main
py 1,696 lines 84 KB
Raw
1 """Deterministic, provider-free public-company stock rule engine.
2
3 ``STOCK_RULE_ENGINE_V1`` consumes only durable records that already passed the
4 Research Readiness boundary. It never imports or invokes a provider adapter.
5 Every scored metric names its V1 rule and evidence, and the input fingerprint
6 makes a stored result reproducible and safely reusable.
7
8 V1 aggregation
9 --------------
10 * Each area is a weighted mean of the available, applicable sub-rules below.
11 Missing optional metrics are omitted rather than scored as zero.
12 * ``overallScore`` is the top-level weighted mean, normalized over scorable and
13 applicable areas. The configured area weights remain fixed and total 100.
14 * ``qualityScore`` uses quality (16), growth (14), balance sheet (9), quarterly
15 trend (9), and governance (5), normalized over scorable members.
16 * ``opportunityScore`` uses valuation (18), catalysts (8), technical (7),
17 current events (7), and sector/macro (3), normalized over scorable members.
18 * ``riskScore`` is 100 minus a normalized resilience score made from balance
19 sheet (30), governance (30), quarterly trend (15), current events (15), and
20 business quality (10). A higher risk score therefore means more risk.
21 * ``confidenceScore`` is independent of every stock score: mandatory/critical
22 coverage 40%, overall coverage 20%, source authority 20%, freshness 20%, less
23 15 points per conflict (maximum 30).
24
25 Decision thresholds before gates/overrides are 85 STRONG_BUY, 75 BUY,
26 65 ACCUMULATE, 50 HOLD, 35 REDUCE, 20 AVOID, otherwise EXIT_REVIEW. Partial
27 analysis is explicitly requested and capped at HOLD. Critical overrides sit
28 above the weighted score and block positive decisions.
29 """
30 from __future__ import annotations
31
32 import asyncio
33 import hashlib
34 import json
35 import math
36 import re
37 from dataclasses import dataclass
38 from datetime import datetime, timedelta, timezone
39 from decimal import Decimal, InvalidOperation
40 from enum import StrEnum
41 from statistics import median, pstdev
42 from typing import Any, Iterable, Mapping, Sequence, TYPE_CHECKING
43
44 if TYPE_CHECKING:
45 from app.news_intelligence import EventImpactFeature, SearchRun
46 from uuid import UUID
47
48 from pydantic import Field
49
50 from app.fact_precedence import FinancialFact, FactSourceTier, fact_source_authority
51 from app.models import (
52 CompanyResearchProfile,
53 EventImpact,
54 MarketPriceObservation,
55 ReliabilityLevel,
56 ResearchBaseModel,
57 ResearchEvent,
58 ResearchEventType,
59 ResearchLifecycleStatus,
60 ShareholdingCategory,
61 ShareholdingSnapshot,
62 SourceClassification,
63 SourceMode,
64 StructuredMarketSnapshotRecord,
65 TimeHorizon,
66 )
67 from app.research_readiness import (
68 ResearchReadinessResult,
69 ResearchRequirementReadiness,
70 ResearchRequirementStatus,
71 ResearchSourceTier,
72 RuleEngineArea,
73 )
74
75
76 STOCK_RULE_ENGINE_VERSION = "STOCK_RULE_ENGINE_V1"
77 CURRENT_NEWS_WINDOW_DAYS = 30
78
79 STOCK_RULE_ENGINE_AREA_WEIGHTS: Mapping[RuleEngineArea, int] = {
80 RuleEngineArea.VALUATION: 18,
81 RuleEngineArea.FUNDAMENTAL_BUSINESS_QUALITY: 16,
82 RuleEngineArea.GROWTH: 14,
83 RuleEngineArea.BALANCE_SHEET: 9,
84 RuleEngineArea.QUARTERLY_EARNINGS_TREND: 9,
85 RuleEngineArea.ORDER_BOOK_CAPACITY_CATALYSTS: 8,
86 RuleEngineArea.PRICE_TECHNICAL: 7,
87 RuleEngineArea.NEWS_GEOPOLITICAL_EVENTS: 7,
88 RuleEngineArea.SHAREHOLDING: 4,
89 RuleEngineArea.MANAGEMENT_GOVERNANCE: 5,
90 RuleEngineArea.SECTOR_MACRO: 3,
91 }
92 if sum(STOCK_RULE_ENGINE_AREA_WEIGHTS.values()) != 100: # pragma: no cover - import guard
93 raise RuntimeError("STOCK_RULE_ENGINE_V1 area weights must total exactly 100")
94
95
96 class AreaScoreStatus(StrEnum):
97 READY_FRESH = "READY_FRESH"
98 READY_STALE = "READY_STALE"
99 PARTIAL = "PARTIAL"
100 CONFLICTING = "CONFLICTING"
101 UNSCORABLE = "UNSCORABLE"
102 UNSUPPORTED = "UNSUPPORTED"
103 NOT_APPLICABLE = "NOT_APPLICABLE"
104
105
106 class DecisionSignal(StrEnum):
107 STRONG_BUY = "STRONG_BUY"
108 BUY = "BUY"
109 ACCUMULATE = "ACCUMULATE"
110 HOLD = "HOLD"
111 REDUCE = "REDUCE"
112 AVOID = "AVOID"
113 EXIT_REVIEW = "EXIT_REVIEW"
114 INSUFFICIENT_DATA = "INSUFFICIENT_DATA"
115
116
117 class ConfidenceLevel(StrEnum):
118 HIGH = "HIGH"
119 MEDIUM = "MEDIUM"
120 LOW = "LOW"
121
122
123 class RiskOverrideSeverity(StrEnum):
124 HIGH = "HIGH"
125 CRITICAL = "CRITICAL"
126
127
128 class RuleMetricResult(ResearchBaseModel):
129 metric: str
130 value: Any = None
131 unit: str | None = None
132 score: float = Field(ge=0, le=100)
133 rule: str
134 configured_subrule_weight: int = Field(default=1, gt=0)
135 applied_weight_pct: float = Field(default=100, ge=0, le=100)
136 source: str
137 source_url: str | None = None
138 as_of: datetime | None = None
139 evidence_references: list[str] = Field(default_factory=list)
140
141
142 class RuleSourceReference(ResearchBaseModel):
143 source_provider: str | None = None
144 source_tier: ResearchSourceTier | None = None
145 source_url: str
146 as_of: datetime | None = None
147 retrieved_at: datetime | None = None
148 evidence_references: list[str] = Field(default_factory=list)
149
150
151 class AreaScoreResult(ResearchBaseModel):
152 area: RuleEngineArea
153 weight: int
154 raw_score: float | None = Field(default=None, ge=0, le=100)
155 weighted_contribution: float = Field(default=0, ge=0, le=100)
156 status: AreaScoreStatus
157 applicable: bool
158 metrics: list[RuleMetricResult] = Field(default_factory=list)
159 positive_factors: list[str] = Field(default_factory=list)
160 negative_factors: list[str] = Field(default_factory=list)
161 evidence_references: list[str] = Field(default_factory=list)
162 source_references: list[RuleSourceReference] = Field(default_factory=list)
163 missing_inputs: list[str] = Field(default_factory=list)
164
165
166 class RiskOverrideResult(ResearchBaseModel):
167 code: str
168 severity: RiskOverrideSeverity
169 effect: str = "BLOCK_BUY"
170 evidence_ids: list[str] = Field(default_factory=list)
171
172
173 class AnalysisEligibility(ResearchBaseModel):
174 full_analysis_allowed: bool
175 partial_analysis_allowed: bool
176 blocking_requirements: list[str] = Field(default_factory=list)
177 reason: str
178
179
180 class StockRuleEngineResult(ResearchBaseModel):
181 rule_engine_version: str = STOCK_RULE_ENGINE_VERSION
182 calculated_at: datetime
183 global_instrument_id: UUID
184 input_as_of: datetime | None = None
185 input_fingerprint: str
186 overall_score: float | None = Field(default=None, ge=0, le=100)
187 quality_score: float | None = Field(default=None, ge=0, le=100)
188 opportunity_score: float | None = Field(default=None, ge=0, le=100)
189 risk_score: float | None = Field(default=None, ge=0, le=100)
190 confidence_score: float = Field(ge=0, le=100)
191 confidence: ConfidenceLevel
192 decision_signal: DecisionSignal
193 partial: bool
194 cache_hit: bool = False
195 eligibility: AnalysisEligibility
196 area_scores: list[AreaScoreResult] = Field(default_factory=list)
197 risk_overrides: list[RiskOverrideResult] = Field(default_factory=list)
198 missing_inputs: list[str] = Field(default_factory=list)
199 evidence_references: list[str] = Field(default_factory=list)
200
201
202 @dataclass(frozen=True)
203 class _Datum:
204 value: Decimal
205 source: str
206 source_url: str | None
207 as_of: datetime | None
208 evidence_id: str
209 unit: str | None = None
210 authority: int = 0
211
212
213 @dataclass(frozen=True)
214 class StockRuleEngineInput:
215 profile: CompanyResearchProfile
216 readiness: ResearchReadinessResult
217 financial_facts: tuple[FinancialFact, ...]
218 structured_snapshots: tuple[StructuredMarketSnapshotRecord, ...]
219 market_prices: tuple[MarketPriceObservation, ...]
220 events: tuple[ResearchEvent, ...]
221 shareholding: tuple[ShareholdingSnapshot, ...]
222 canonical_metadata: Mapping[str, Any]
223 evaluated_at: datetime
224 news_features: tuple[EventImpactFeature, ...] = ()
225 news_search_run: SearchRun | None = None
226
227
228 class StockRuleEngineEligibilityPolicy:
229 """One deterministic gate shared by the API and UI action contract."""
230
231 CRITICAL_REQUIREMENTS = (
232 "VALUATION_INPUTS",
233 "BUSINESS_QUALITY_FACTS",
234 "BALANCE_SHEET_FACTS",
235 "QUARTERLY_FINANCIALS",
236 "LATEST_PRICE",
237 )
238 FULL_STATUSES = frozenset(
239 {ResearchRequirementStatus.READY_FRESH, ResearchRequirementStatus.READY_STALE}
240 )
241 PARTIAL_STATUSES = frozenset(
242 {
243 ResearchRequirementStatus.READY_FRESH,
244 ResearchRequirementStatus.READY_STALE,
245 ResearchRequirementStatus.PARTIAL,
246 }
247 )
248
249 def evaluate(self, readiness: ResearchReadinessResult) -> AnalysisEligibility:
250 by_id = {item.requirement_id: item for item in readiness.requirements}
251 mandatory_blocking = sorted(
252 item.requirement_id
253 for item in readiness.requirements
254 if item.mandatory and item.requirement_id != 'CURRENT_NEWS' and item.status not in self.FULL_STATUSES and item.status != ResearchRequirementStatus.NOT_APPLICABLE
255 )
256 critical = [by_id.get(key) for key in self.CRITICAL_REQUIREMENTS]
257 critical_blocking = sorted(
258 key
259 for key, item in zip(self.CRITICAL_REQUIREMENTS, critical)
260 if item is None or (item.status not in self.FULL_STATUSES and item.status != ResearchRequirementStatus.NOT_APPLICABLE)
261 )
262 full = (
263 readiness.critical_completeness_pct >= 80
264 and not mandatory_blocking
265 and not critical_blocking
266 )
267 usable_critical = sum(
268 1 for item in critical if item is not None and item.status in self.PARTIAL_STATUSES
269 )
270 latest = by_id.get("LATEST_PRICE")
271 conflict = any(
272 item is not None and item.status == ResearchRequirementStatus.CONFLICTING
273 for item in critical
274 )
275 partial = (
276 not full
277 and readiness.critical_completeness_pct >= 50
278 and usable_critical >= 3
279 and latest is not None
280 and latest.status in self.PARTIAL_STATUSES
281 and not conflict
282 )
283 blockers = sorted(set(mandatory_blocking + critical_blocking))
284 reason = (
285 "FULL_CRITICAL_AND_MANDATORY_INPUTS_AVAILABLE"
286 if full
287 else "MINIMUM_SAFE_PARTIAL_GATE_AVAILABLE"
288 if partial
289 else "MINIMUM_SAFE_CRITICAL_INPUTS_UNAVAILABLE"
290 )
291 return AnalysisEligibility(
292 full_analysis_allowed=full,
293 partial_analysis_allowed=partial,
294 blocking_requirements=blockers,
295 reason=reason,
296 )
297
298
299 class StockRuleEngineInputAdapter:
300 """Loads existing durable public-company data; it has no provider handles."""
301
302 def __init__(self, repository, readiness_adapter) -> None:
303 self.repository = repository
304 self.readiness_adapter = readiness_adapter
305
306 async def load(
307 self,
308 profile: CompanyResearchProfile,
309 readiness: ResearchReadinessResult,
310 *,
311 now: datetime | None = None,
312 ) -> StockRuleEngineInput:
313 instrument_id = profile.instrument_id
314 facts, snapshots, prices = await asyncio.gather(
315 self.repository.financial_facts_for_instruments({instrument_id}),
316 self.repository.structured_market_snapshots_for_instruments({instrument_id}),
317 self.repository.market_price_observations_for_instruments({instrument_id}),
318 )
319 evaluated_at = _aware(now or datetime.now(timezone.utc))
320 from app.news_intelligence import EventImpactFeature, SearchRun
321 news_loader = getattr(self.repository, 'news_records_for', None)
322 features = news_loader(instrument_id, EventImpactFeature, as_of=evaluated_at) if callable(news_loader) else []
323 runs = news_loader(instrument_id, SearchRun, as_of=evaluated_at) if callable(news_loader) else []
324 return StockRuleEngineInput(
325 profile=profile,
326 readiness=readiness,
327 financial_facts=tuple(facts.get(instrument_id, ())),
328 structured_snapshots=tuple(snapshots.get(instrument_id, ())),
329 market_prices=tuple(prices.get(instrument_id, ())),
330 events=tuple(
331 self.repository.events_for(instrument_id, source_mode=SourceMode.REAL)
332 ),
333 shareholding=tuple(self.repository.shareholding_for(instrument_id, limit=8)),
334 canonical_metadata=self.readiness_adapter.canonical_metadata_for(instrument_id),
335 evaluated_at=evaluated_at,
336 news_features=tuple(features), news_search_run=runs[-1] if runs else None,
337 )
338
339
340 class StockRuleEngineService:
341 """Application service for deterministic scoring and exact-input caching."""
342
343 def __init__(self, repository, readiness_adapter) -> None:
344 self.repository = repository
345 self.input_adapter = StockRuleEngineInputAdapter(repository, readiness_adapter)
346 self.engine = StockRuleEngineV1()
347 self.eligibility_policy = StockRuleEngineEligibilityPolicy()
348
349 async def analyze(
350 self,
351 profile: CompanyResearchProfile,
352 readiness: ResearchReadinessResult,
353 *,
354 allow_partial: bool,
355 now: datetime | None = None,
356 ) -> StockRuleEngineResult:
357 inputs = await self.input_adapter.load(profile, readiness, now=now)
358 fingerprint = self.engine.input_fingerprint(inputs, allow_partial=allow_partial)
359 cached = await self.repository.stock_rule_engine_result(
360 profile.instrument_id, STOCK_RULE_ENGINE_VERSION, fingerprint
361 )
362 if cached is not None:
363 return StockRuleEngineResult.model_validate(cached).model_copy(
364 update={"cache_hit": True}
365 )
366 result = self.engine.evaluate(inputs, allow_partial=allow_partial, fingerprint=fingerprint)
367 await self.repository.persist_stock_rule_engine_result(
368 result.model_dump(mode="json", by_alias=False)
369 )
370 return result
371
372
373 class StockRuleEngineV1:
374 """Versioned sub-rule evaluators over one immutable durable input snapshot."""
375
376 def __init__(self) -> None:
377 self.eligibility_policy = StockRuleEngineEligibilityPolicy()
378
379 def input_fingerprint(self, value: StockRuleEngineInput, *, allow_partial: bool) -> str:
380 payload = {
381 "version": STOCK_RULE_ENGINE_VERSION,
382 # Fingerprint schema for the V1 payload. This preserves exact-cache
383 # safety when explainability fields evolve before a new score rule.
384 "fingerprintContract": "STOCK_RULE_ENGINE_V1_INPUT_2_NEWS",
385 "newsFeatures": [f.model_dump(mode='json') for f in sorted(value.news_features,key=lambda f:str(f.feature_id))],
386 "newsSearch": value.news_search_run.model_dump(mode='json') if value.news_search_run else None,
387 "newsEvaluationDate": value.evaluated_at.isoformat() if value.news_features or value.news_search_run else None,
388 "analysisMode": "PARTIAL_ALLOWED" if allow_partial else "FULL_REQUIRED",
389 # Aging current-news eligibility changes at UTC day boundaries.
390 "evaluationDate": value.evaluated_at.date().isoformat(),
391 "globalInstrumentId": str(value.profile.instrument_id),
392 "profile": {
393 "country": value.profile.country,
394 "exchange": value.profile.exchange,
395 "mic": value.profile.mic,
396 "ticker": value.profile.ticker,
397 "sector": _sector(value),
398 },
399 "readiness": [
400 {
401 "id": item.requirement_id,
402 "status": item.status.value,
403 "coverage": item.coverage_pct,
404 "criticalCoverage": item.critical_coverage_pct,
405 "source": item.source,
406 "sourceTier": item.source_tier.value if item.source_tier else None,
407 "sourceUrl": item.source_url,
408 "asOf": _iso(item.as_of),
409 "evidence": sorted(item.evidence_ids),
410 "missing": sorted(item.missing_input_ids),
411 "conflict": item.conflict_reason,
412 }
413 for item in sorted(value.readiness.requirements, key=lambda item: item.requirement_id)
414 ],
415 "financialFacts": [
416 {
417 "metric": fact.key.metric,
418 "periodEnd": fact.key.period_end,
419 "periodType": fact.key.period_type,
420 "basis": fact.key.reporting_basis,
421 "value": str(fact.value.value),
422 "unit": fact.value.unit,
423 "source": fact.source_provider,
424 "sourceIdentity": fact.source_identity,
425 "sourceTier": int(fact.source_tier),
426 "sourceUrl": fact.value.source_url,
427 "asOf": _iso(fact.value.as_of_date or _period_datetime(fact.key.period_end)),
428 }
429 for fact in sorted(
430 value.financial_facts,
431 key=lambda item: (
432 _metric_key(item.key.metric), item.key.period_type,
433 item.key.period_end or "", item.key.reporting_basis or "",
434 ),
435 )
436 if fact.source_mode == SourceMode.REAL
437 ],
438 "structured": [
439 {
440 "provider": record.provider,
441 "providerId": record.provider_instrument_id,
442 "marketAsOf": _iso(record.market_as_of),
443 "facts": {
444 key: {
445 "value": str(item.value),
446 "asOf": _iso(item.as_of_date),
447 "source": item.source_name,
448 "url": item.source_url,
449 }
450 for key, item in sorted(record.snapshot.facts.items())
451 if item.value is not None
452 },
453 }
454 for record in sorted(value.structured_snapshots, key=lambda item: item.provider)
455 ],
456 "prices": [
457 {
458 "at": _iso(item.observed_at),
459 "price": str(item.price),
460 "provider": item.provider,
461 }
462 for item in _usable_prices(value.market_prices)
463 ],
464 "events": [
465 {
466 "id": str(item.event_id),
467 "type": _enum_text(item.event_type),
468 "date": _iso(item.event_date or item.published_at),
469 "title": item.title,
470 "summary": item.summary,
471 "impact": _enum_text(item.impact),
472 "horizon": _enum_text(item.time_horizon),
473 "confidence": item.confidence,
474 "status": _enum_text(item.status),
475 "classification": _enum_text(item.source_classification),
476 "reliability": _enum_text(item.reliability),
477 "sourceUrl": item.source_url,
478 "monetary": str(item.monetary_value) if item.monetary_value is not None else None,
479 "percentage": str(item.percentage_value) if item.percentage_value is not None else None,
480 "capacity": str(item.capacity_value) if item.capacity_value is not None else None,
481 "counterparty": item.counterparty,
482 "customer": item.customer,
483 "location": item.location,
484 }
485 for item in sorted(value.events, key=lambda item: str(item.event_id))
486 ],
487 "shareholding": [
488 {
489 "id": str(item.id),
490 "periodEnd": _iso(item.period_end),
491 "provider": item.source_provider,
492 "url": item.source_url,
493 "values": sorted(
494 (_enum_text(entry.category), str(entry.percentage), entry.metric_basis)
495 for entry in item.values
496 ),
497 }
498 for item in sorted(value.shareholding, key=lambda item: item.period_end)
499 ],
500 }
501 encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
502 return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
503
504 def evaluate(
505 self,
506 value: StockRuleEngineInput,
507 *,
508 allow_partial: bool,
509 fingerprint: str | None = None,
510 ) -> StockRuleEngineResult:
511 eligibility = self.eligibility_policy.evaluate(value.readiness)
512 areas = [
513 self._valuation(value),
514 self._quality(value),
515 self._growth(value),
516 self._balance_sheet(value),
517 self._quarterly(value),
518 self._catalysts(value),
519 self._technical(value),
520 self._news(value),
521 self._shareholding(value),
522 self._governance(value),
523 self._sector_macro(value),
524 ]
525 scorable = [item for item in areas if item.applicable and item.raw_score is not None]
526 raw_overall = _area_weighted_score(scorable)
527 quality = _area_weighted_score(
528 scorable,
529 areas={
530 RuleEngineArea.FUNDAMENTAL_BUSINESS_QUALITY,
531 RuleEngineArea.GROWTH,
532 RuleEngineArea.BALANCE_SHEET,
533 RuleEngineArea.QUARTERLY_EARNINGS_TREND,
534 RuleEngineArea.MANAGEMENT_GOVERNANCE,
535 },
536 )
537 opportunity = _area_weighted_score(
538 scorable,
539 areas={
540 RuleEngineArea.VALUATION,
541 RuleEngineArea.ORDER_BOOK_CAPACITY_CATALYSTS,
542 RuleEngineArea.PRICE_TECHNICAL,
543 RuleEngineArea.NEWS_GEOPOLITICAL_EVENTS,
544 RuleEngineArea.SECTOR_MACRO,
545 },
546 )
547 risk_resilience = _custom_area_score(
548 scorable,
549 {
550 RuleEngineArea.BALANCE_SHEET: 30,
551 RuleEngineArea.MANAGEMENT_GOVERNANCE: 30,
552 RuleEngineArea.QUARTERLY_EARNINGS_TREND: 15,
553 RuleEngineArea.NEWS_GEOPOLITICAL_EVENTS: 15,
554 RuleEngineArea.FUNDAMENTAL_BUSINESS_QUALITY: 10,
555 },
556 )
557 risk = _round_score(100 - risk_resilience) if risk_resilience is not None else None
558 confidence_score = self._confidence(value.readiness)
559 confidence = _confidence_level(confidence_score)
560 overrides = self._risk_overrides(value)
561
562 permitted = eligibility.full_analysis_allowed or (
563 allow_partial and eligibility.partial_analysis_allowed
564 )
565 partial = not eligibility.full_analysis_allowed
566 overall = raw_overall if permitted else None
567 decision = (
568 self._decision(
569 raw_overall,
570 value.readiness,
571 confidence,
572 partial=partial,
573 overrides=overrides,
574 )
575 if permitted
576 else DecisionSignal.INSUFFICIENT_DATA
577 )
578 missing = sorted(
579 set(eligibility.blocking_requirements)
580 | {missing for area in areas for missing in area.missing_inputs}
581 )
582 evidence = sorted(
583 {reference for area in areas for reference in area.evidence_references}
584 | {reference for override in overrides for reference in override.evidence_ids}
585 )
586 return StockRuleEngineResult(
587 calculated_at=value.evaluated_at,
588 global_instrument_id=value.profile.instrument_id,
589 input_as_of=_input_as_of(value),
590 input_fingerprint=fingerprint or self.input_fingerprint(value, allow_partial=allow_partial),
591 overall_score=overall,
592 quality_score=quality if permitted else None,
593 opportunity_score=opportunity if permitted else None,
594 risk_score=risk if permitted else None,
595 confidence_score=confidence_score,
596 confidence=confidence,
597 decision_signal=decision,
598 partial=partial,
599 eligibility=eligibility,
600 area_scores=areas,
601 risk_overrides=overrides,
602 missing_inputs=missing,
603 evidence_references=evidence,
604 )
605
606 def _valuation(self, value: StockRuleEngineInput) -> AreaScoreResult:
607 metrics: list[tuple[RuleMetricResult, int]] = []
608 structured = _structured_data(value)
609 from app.valuation_evidence import materialize_valuation
610 for name, fact in materialize_valuation(value.structured_snapshots,value.market_prices,now=value.evaluated_at).items():
611 structured[_metric_key(name)] = _Datum(Decimal(str(fact.value)),fact.source_name,fact.source_url,
612 fact.as_of_date,'derived:'+name,'RATIO',0)
613 pe = _pick(structured, "trailingpe", "pe", "pricetoearnings")
614 forward_pe = _pick(structured, "forwardpe")
615 pb = _pick(structured, "pricetobook", "pb")
616 ev_ebitda = _pick(structured, "evtoebitda", "enterprisevaluetoebitda")
617 peg = _pick(structured, "pegratio", "peg")
618 market_cap = _pick(structured, "marketcap")
619 fcf = _pick(structured, "freecashflow") or _latest_fact(value, ("free_cash_flow",))
620 financial = _is_financial(value)
621 if pe and pe.value > 0:
622 metrics.append((_metric_result("TRAILING_PE", pe, _lower_better(pe.value, ((8, 92), (15, 78), (25, 58), (40, 32), (60, 12))), "TRAILING_PE_V1", "RATIO"), 25))
623 earnings_yield = _derived_datum(Decimal("100") / pe.value, "PERCENT", pe, "derived:earnings-yield")
624 metrics.append((_metric_result("EARNINGS_YIELD", earnings_yield, _higher_better(earnings_yield.value, ((1.5, 10), (2.5, 30), (4, 55), (6.5, 78), (10, 92))), "EARNINGS_YIELD_V1", "PERCENT"), 10))
625 if forward_pe and forward_pe.value > 0:
626 metrics.append((_metric_result("FORWARD_PE", forward_pe, _lower_better(forward_pe.value, ((8, 94), (15, 80), (25, 58), (40, 30), (60, 10))), "FORWARD_PE_V1", "RATIO"), 15))
627 if pb and pb.value > 0:
628 points = ((0.8, 94), (1.5, 82), (3, 60), (5, 38), (8, 15)) if financial else ((1, 88), (2, 76), (4, 56), (8, 30), (12, 12))
629 metrics.append((_metric_result("PRICE_TO_BOOK", pb, _lower_better(pb.value, points), "PRICE_TO_BOOK_FINANCIAL_V1" if financial else "PRICE_TO_BOOK_GENERAL_V1", "RATIO"), 18 if financial else 8))
630 if ev_ebitda and ev_ebitda.value > 0 and not financial:
631 metrics.append((_metric_result("EV_EBITDA", ev_ebitda, _lower_better(ev_ebitda.value, ((5, 94), (8, 80), (12, 62), (20, 35), (30, 12))), "EV_EBITDA_NON_FINANCIAL_V1", "RATIO"), 15))
632 if peg and peg.value > 0:
633 metrics.append((_metric_result("PEG", peg, _lower_better(peg.value, ((0.5, 90), (1, 82), (1.5, 65), (2.5, 38), (4, 15))), "PEG_V1", "RATIO"), 10))
634 if fcf and market_cap and market_cap.value > 0:
635 fcf_yield = _derived_datum(fcf.value / market_cap.value * 100, "PERCENT", fcf, "derived:fcf-yield", market_cap)
636 metrics.append((_metric_result("FCF_YIELD", fcf_yield, _higher_better(fcf_yield.value, ((-5, 5), (0, 25), (2, 48), (5, 72), (8, 90))), "FCF_YIELD_V1", "PERCENT"), 15))
637 missing = [] if metrics else ["VALUATION_BASIS"]
638 return self._finish(value, RuleEngineArea.VALUATION, metrics, missing)
639
640 def _quality(self, value: StockRuleEngineInput) -> AreaScoreResult:
641 metrics: list[tuple[RuleMetricResult, int]] = []
642 structured = _structured_data(value)
643 roe = _pick(structured, "roe", "returnonequity") or _latest_fact(value, ("roe", "return_on_equity"))
644 roce = _pick(structured, "roce", "returnoncapitalemployed") or _latest_fact(value, ("roce", "return_on_capital_employed"))
645 operating_margin = _pick(structured, "operatingmargin") or _latest_fact(value, ("operating_margin", "ebitda_margin"))
646 net_margin = _pick(structured, "profitmargin", "netmargin") or _latest_fact(value, ("profit_margin", "net_margin"))
647 for name, datum, weight, rule in (
648 ("ROE", roe, 20, "ROE_V1"),
649 ("ROCE", roce if not _is_financial(value) else None, 20, "ROCE_NON_FINANCIAL_V1"),
650 ("OPERATING_MARGIN", operating_margin, 12, "OPERATING_MARGIN_V1"),
651 ("NET_MARGIN", net_margin, 10, "NET_MARGIN_V1"),
652 ):
653 if datum:
654 normalized = _as_percent(datum.value)
655 normalized_datum = _derived_datum(normalized, "PERCENT", datum, f"normalized:{name.lower()}")
656 metrics.append((_metric_result(name, normalized_datum, _higher_better(normalized, ((0, 15), (5, 38), (10, 58), (18, 78), (28, 94))), rule, "PERCENT"), weight))
657
658 annual_pat = _fact_series(value, ("pat", "net_income", "net_profit"), "ANNUAL")
659 annual_revenue = _fact_series(value, ("revenue", "total_revenue"), "ANNUAL")
660 margins = _aligned_ratios(annual_pat, annual_revenue, multiplier=Decimal("100"))
661 if len(margins) >= 3:
662 stability = Decimal(str(pstdev(float(item) for item in margins[-5:])))
663 datum = _derived_from_many(Decimal("100") - stability, "INDEX", [*annual_pat[-5:], *annual_revenue[-5:]], "derived:margin-stability")
664 metrics.append((_metric_result("MARGIN_STABILITY", datum, _lower_better(stability, ((2, 92), (5, 78), (10, 55), (20, 25), (35, 8))), "MARGIN_STABILITY_STDDEV_V1", "PERCENT_STDDEV", display_value=stability), 12))
665 ocf = _latest_fact(value, ("operating_cash_flow", "cash_flow_from_operating_activities"), period_type="ANNUAL")
666 pat = annual_pat[-1] if annual_pat else _latest_fact(value, ("pat", "net_income", "net_profit"), period_type="ANNUAL")
667 if ocf and pat and pat.value != 0:
668 conversion = ocf.value / abs(pat.value)
669 datum = _derived_datum(conversion, "RATIO", ocf, "derived:cash-conversion", pat)
670 metrics.append((_metric_result("CASH_CONVERSION", datum, _higher_better(conversion, ((0, 8), (0.5, 40), (0.8, 65), (1, 82), (1.3, 94))), "OPERATING_CASH_TO_PAT_V1", "RATIO"), 16))
671 fcf = _pick(structured, "freecashflow") or _latest_fact(value, ("free_cash_flow",), period_type="ANNUAL")
672 if fcf and pat and pat.value != 0:
673 quality = fcf.value / abs(pat.value)
674 datum = _derived_datum(quality, "RATIO", fcf, "derived:fcf-quality", pat)
675 metrics.append((_metric_result("FCF_QUALITY", datum, _higher_better(quality, ((-0.5, 5), (0, 25), (0.5, 55), (0.8, 75), (1.1, 90))), "FCF_TO_PAT_V1", "RATIO"), 10))
676 if len(annual_pat) >= 3:
677 consistency = Decimal(sum(1 for item in annual_pat[-5:] if item.value > 0)) / Decimal(len(annual_pat[-5:])) * 100
678 datum = _derived_from_many(consistency, "PERCENT", annual_pat[-5:], "derived:profitability-consistency")
679 metrics.append((_metric_result("PROFITABILITY_CONSISTENCY", datum, float(consistency), "POSITIVE_PAT_PERIOD_SHARE_V1", "PERCENT"), 16))
680 missing = [] if metrics else ["QUALITY_FACTS"]
681 return self._finish(value, RuleEngineArea.FUNDAMENTAL_BUSINESS_QUALITY, metrics, missing)
682
683 def _growth(self, value: StockRuleEngineInput) -> AreaScoreResult:
684 metrics: list[tuple[RuleMetricResult, int]] = []
685 annual_revenue = _fact_series(value, ("revenue", "total_revenue"), "ANNUAL")
686 annual_earnings = _fact_series(value, ("pat", "net_income", "net_profit"), "ANNUAL")
687 if not annual_earnings:
688 annual_earnings = _fact_series(value, ("eps",), "ANNUAL")
689 for name, series, weight in (
690 ("REVENUE_CAGR", annual_revenue, 27),
691 ("EARNINGS_CAGR", annual_earnings, 27),
692 ):
693 cagr = _series_cagr(series)
694 if cagr is not None:
695 datum = _derived_from_many(cagr, "PERCENT", series, f"derived:{name.lower()}")
696 metrics.append((_metric_result(name, datum, _growth_score(cagr), f"{name}_MULTI_YEAR_V1", "PERCENT"), weight))
697 quarterly_revenue = _fact_series(value, ("revenue", "total_revenue"), "QUARTERLY")
698 quarterly_earnings = _fact_series(value, ("pat", "net_income", "net_profit"), "QUARTERLY")
699 if not quarterly_earnings:
700 quarterly_earnings = _fact_series(value, ("eps",), "QUARTERLY")
701 for name, series, weight in (
702 ("REVENUE_YOY", quarterly_revenue, 15),
703 ("EARNINGS_YOY", quarterly_earnings, 15),
704 ):
705 comparison = _comparable_yoy(series)
706 if comparison:
707 change, latest, prior = comparison
708 datum = _derived_datum(change, "PERCENT", latest, f"derived:{name.lower()}", prior)
709 metrics.append((_metric_result(name, datum, _growth_score(change), f"{name}_COMPARABLE_QUARTER_V1", "PERCENT"), weight))
710 qoq = _sequential_change(quarterly_revenue)
711 if qoq:
712 change, latest, prior = qoq
713 datum = _derived_datum(change, "PERCENT", latest, "derived:revenue-qoq", prior)
714 metrics.append((_metric_result("RECENT_REVENUE_QOQ", datum, _growth_score(change), "RECENT_QOQ_LIMITED_WEIGHT_V1", "PERCENT"), 6))
715 consistency = _growth_consistency(annual_revenue, annual_earnings)
716 if consistency:
717 score, source_data = consistency
718 datum = _derived_from_many(score, "PERCENT", source_data, "derived:growth-consistency")
719 metrics.append((_metric_result("MULTI_YEAR_GROWTH_CONSISTENCY", datum, float(score), "POSITIVE_ANNUAL_CHANGE_SHARE_V1", "PERCENT"), 10))
720 return self._finish(value, RuleEngineArea.GROWTH, metrics, [] if metrics else ["MULTI_PERIOD_GROWTH_HISTORY"])
721
722 def _balance_sheet(self, value: StockRuleEngineInput) -> AreaScoreResult:
723 metrics: list[tuple[RuleMetricResult, int]] = []
724 if _is_financial(value):
725 capital = _latest_fact(value, ("capital_adequacy", "capital_adequacy_ratio"))
726 gross_npa = _latest_fact(value, ("gross_npa", "gross_npa_ratio"))
727 net_npa = _latest_fact(value, ("net_npa", "net_npa_ratio"))
728 if capital:
729 pct = _as_percent(capital.value)
730 metrics.append((_metric_result("CAPITAL_ADEQUACY", _derived_datum(pct, "PERCENT", capital, "normalized:capital-adequacy"), _higher_better(pct, ((9, 20), (12, 50), (15, 72), (18, 90))), "FINANCIAL_CAPITAL_ADEQUACY_V1", "PERCENT"), 50))
731 for name, datum, weight in (("GROSS_NPA", gross_npa, 25), ("NET_NPA", net_npa, 25)):
732 if datum:
733 pct = _as_percent(datum.value)
734 metrics.append((_metric_result(name, _derived_datum(pct, "PERCENT", datum, f"normalized:{name.lower()}"), _lower_better(pct, ((1, 92), (2, 78), (4, 52), (7, 25), (12, 8))), f"{name}_FINANCIAL_V1", "PERCENT"), weight))
735 return self._finish(value, RuleEngineArea.BALANCE_SHEET, metrics, [] if metrics else ["FINANCIAL_SECTOR_CAPITAL_OR_ASSET_QUALITY"], applicable=True)
736
737 structured = _structured_data(value)
738 debt_equity = _pick(structured, "debttoequity")
739 debt = _pick(structured, "totaldebt") or _latest_fact(value, ("debt_or_borrowings", "total_debt"), period_type="ANNUAL")
740 equity = _latest_fact(value, ("equity", "total_equity"), period_type="ANNUAL")
741 cash = _pick(structured, "totalcash") or _latest_fact(value, ("cash_and_cash_equivalents", "cash_and_equivalents"), period_type="ANNUAL")
742 if debt_equity:
743 de_pct = debt_equity.value if abs(debt_equity.value) > 5 else debt_equity.value * 100
744 datum = _derived_datum(de_pct, "PERCENT", debt_equity, "normalized:debt-equity")
745 elif debt and equity and equity.value > 0:
746 de_pct = debt.value / equity.value * 100
747 datum = _derived_datum(de_pct, "PERCENT", debt, "derived:debt-equity", equity)
748 else:
749 datum = None
750 if datum:
751 metrics.append((_metric_result("DEBT_TO_EQUITY", datum, _lower_better(datum.value, ((0, 95), (30, 85), (60, 68), (100, 45), (200, 18))), "INDUSTRIAL_DEBT_TO_EQUITY_V1", "PERCENT"), 28))
752 if debt and cash and equity and equity.value > 0:
753 net_debt_equity = (debt.value - cash.value) / equity.value * 100
754 nd = _derived_datum(net_debt_equity, "PERCENT", debt, "derived:net-debt-equity", cash, equity)
755 metrics.append((_metric_result("NET_DEBT_TO_EQUITY", nd, _lower_better(net_debt_equity, ((-20, 98), (0, 90), (30, 75), (80, 48), (150, 18))), "NET_DEBT_TO_EQUITY_V1", "PERCENT"), 18))
756 ebitda = _latest_fact(value, ("ebitda", "ebit", "operating_income"), period_type="ANNUAL")
757 finance_cost = _latest_fact(value, ("finance_cost", "interest_expense"), period_type="ANNUAL")
758 if ebitda and finance_cost and finance_cost.value > 0:
759 coverage = ebitda.value / finance_cost.value
760 cov = _derived_datum(coverage, "RATIO", ebitda, "derived:interest-coverage", finance_cost)
761 metrics.append((_metric_result("INTEREST_COVERAGE", cov, _higher_better(coverage, ((0.5, 5), (1, 18), (2, 45), (4, 70), (8, 92))), "EBITDA_INTEREST_COVERAGE_V1", "RATIO"), 24))
762 current_assets = _latest_fact(value, ("current_assets",), period_type="ANNUAL")
763 current_liabilities = _latest_fact(value, ("current_liabilities",), period_type="ANNUAL")
764 if current_assets and current_liabilities and current_liabilities.value > 0:
765 current_ratio = current_assets.value / current_liabilities.value
766 current = _derived_datum(current_ratio, "RATIO", current_assets, "derived:current-ratio", current_liabilities)
767 metrics.append((_metric_result("CURRENT_RATIO", current, _higher_better(current_ratio, ((0.5, 8), (0.9, 35), (1.2, 65), (1.8, 88), (3, 82))), "CURRENT_RATIO_NON_FINANCIAL_V1", "RATIO"), 15))
768 debt_series = _fact_series(value, ("debt_or_borrowings", "total_debt"), "ANNUAL")
769 if len(debt_series) >= 2 and debt_series[-2].value > 0:
770 trend = (debt_series[-1].value / debt_series[-2].value - 1) * 100
771 debt_trend = _derived_datum(trend, "PERCENT", debt_series[-1], "derived:debt-trend", debt_series[-2])
772 metrics.append((_metric_result("DEBT_TREND", debt_trend, _lower_better(trend, ((-20, 95), (0, 78), (10, 58), (30, 30), (60, 8))), "ANNUAL_DEBT_TREND_V1", "PERCENT"), 15))
773 return self._finish(value, RuleEngineArea.BALANCE_SHEET, metrics, [] if metrics else ["BALANCE_SHEET_LEVERAGE_OR_LIQUIDITY"])
774
775 def _quarterly(self, value: StockRuleEngineInput) -> AreaScoreResult:
776 metrics: list[tuple[RuleMetricResult, int]] = []
777 revenue = _fact_series(value, ("revenue", "total_revenue"), "QUARTERLY")
778 earnings = _fact_series(value, ("pat", "net_income", "net_profit"), "QUARTERLY")
779 eps = _fact_series(value, ("eps",), "QUARTERLY")
780 ebitda = _fact_series(value, ("ebitda", "operating_income", "operating_profit"), "QUARTERLY")
781 for name, series, weight in (
782 ("REVENUE_YOY", revenue, 25),
783 ("PAT_YOY", earnings, 25),
784 ("EPS_YOY", eps, 15),
785 ):
786 comparison = _comparable_yoy(series)
787 if comparison:
788 change, latest, prior = comparison
789 datum = _derived_datum(change, "PERCENT", latest, f"derived:quarterly-{name.lower()}", prior)
790 metrics.append((_metric_result(name, datum, _growth_score(change), f"COMPARABLE_QUARTER_{name}_V1", "PERCENT"), weight))
791 for name, series, weight in (("REVENUE_QOQ", revenue, 5), ("PAT_QOQ", earnings, 5)):
792 comparison = _sequential_change(series)
793 if comparison:
794 change, latest, prior = comparison
795 datum = _derived_datum(change, "PERCENT", latest, f"derived:quarterly-{name.lower()}", prior)
796 metrics.append((_metric_result(name, datum, _growth_score(change), f"SEQUENTIAL_{name}_LIMITED_WEIGHT_V1", "PERCENT"), weight))
797 margins = _aligned_ratios(ebitda, revenue, multiplier=Decimal("100"))
798 if len(margins) >= 2:
799 change = margins[-1] - margins[-2]
800 datum = _derived_from_many(change, "PERCENTAGE_POINTS", [*ebitda[-2:], *revenue[-2:]], "derived:quarterly-margin-trend")
801 metrics.append((_metric_result("OPERATING_MARGIN_TREND", datum, _higher_better(change, ((-8, 8), (-3, 30), (0, 55), (2, 75), (5, 92))), "QUARTERLY_MARGIN_TREND_V1", "PERCENTAGE_POINTS"), 15))
802 if len(earnings) >= 3:
803 consistency = Decimal(sum(1 for item in earnings[-4:] if item.value > 0)) / Decimal(len(earnings[-4:])) * 100
804 datum = _derived_from_many(consistency, "PERCENT", earnings[-4:], "derived:quarterly-consistency")
805 metrics.append((_metric_result("EARNINGS_CONSISTENCY", datum, float(consistency), "POSITIVE_QUARTER_SHARE_V1", "PERCENT"), 10))
806 return self._finish(value, RuleEngineArea.QUARTERLY_EARNINGS_TREND, metrics, [] if metrics else ["COMPARABLE_QUARTERLY_RESULTS"])
807
808 def _catalysts(self, value: StockRuleEngineInput) -> AreaScoreResult:
809 relevant = [event for event in value.events if _material_catalyst(event)]
810 metrics = [(_event_metric(event, "MATERIAL_CATALYST_EVENT_V1"), 1) for event in relevant]
811 return self._finish(value, RuleEngineArea.ORDER_BOOK_CAPACITY_CATALYSTS, metrics, [] if metrics else ["MATERIAL_ISSUER_RELEVANT_CATALYST_EVIDENCE"])
812
813 def _technical(self, value: StockRuleEngineInput) -> AreaScoreResult:
814 prices = _usable_prices(value.market_prices)
815 if len(prices) < 50:
816 return self._finish(value, RuleEngineArea.PRICE_TECHNICAL, [], ["FIFTY_OBSERVATION_TECHNICAL_BASIS"])
817 latest = prices[-1]
818 latest_datum = _price_datum(latest)
819 metrics: list[tuple[RuleMetricResult, int]] = []
820 basis50 = Decimal(str(median(float(item.price) for item in prices[-50:])))
821 relative50 = (latest.price / basis50 - 1) * 100
822 rel50 = _derived_datum(relative50, "PERCENT", latest_datum, "derived:price-vs-median-50")
823 metrics.append((_metric_result("PRICE_VS_50_OBSERVATION_MEDIAN", rel50, _trend_score(relative50), "PRICE_VS_50_OBSERVATION_MEDIAN_V1", "PERCENT"), 35))
824 if len(prices) >= 150:
825 basis150 = Decimal(str(median(float(item.price) for item in prices[-150:])))
826 relative150 = (latest.price / basis150 - 1) * 100
827 rel150 = _derived_datum(relative150, "PERCENT", latest_datum, "derived:price-vs-median-150")
828 metrics.append((_metric_result("PRICE_VS_150_OBSERVATION_MEDIAN", rel150, _trend_score(relative150), "PRICE_VS_150_OBSERVATION_MEDIAN_V1", "PERCENT"), 35))
829 window = prices[-150:] if len(prices) >= 150 else prices[-50:]
830 peak = max(item.price for item in window)
831 drawdown = (latest.price / peak - 1) * 100
832 drawdown_datum = _derived_datum(drawdown, "PERCENT", latest_datum, "derived:drawdown")
833 metrics.append((_metric_result("DRAWDOWN_FROM_OBSERVATION_PEAK", drawdown_datum, _higher_better(drawdown, ((-50, 5), (-30, 25), (-15, 52), (-5, 75), (0, 90))), "PRICE_DRAWDOWN_V1", "PERCENT"), 20))
834 start = _price_datum(window[0])
835 trend = (latest.price / window[0].price - 1) * 100
836 trend_datum = _derived_datum(trend, "PERCENT", latest_datum, "derived:price-trend", start)
837 metrics.append((_metric_result("OBSERVATION_WINDOW_TREND", trend_datum, _trend_score(trend), "DURABLE_PRICE_TREND_V1", "PERCENT"), 10))
838 return self._finish(value, RuleEngineArea.PRICE_TECHNICAL, metrics, [])
839
840 def _news(self, value: StockRuleEngineInput) -> AreaScoreResult:
841 from app.news_intelligence import aggregate_impact, search_state
842 normalized = aggregate_impact(value.news_features, value.evaluated_at)
843 state = search_state(value.news_search_run, value.evaluated_at)
844 if normalized is not None or state == 'READY_NO_EVENTS':
845 impact = normalized if normalized is not None else 0
846 refs = sorted(str(f.feature_id) for f in value.news_features) if normalized is not None else ['search-run:'+str(value.news_search_run.run_id)]
847 metric = RuleMetricResult(metric='COMPANY_NEWS_IMPACT',value=impact,unit='IMPACT_MINUS100_PLUS100',score=50+impact/2,
848 rule='COMPANY_EXPOSURE_IMPACT_V2',source='PERSISTED_NEWS_INTELLIGENCE',evidence_references=refs)
849 result=self._finish(value,RuleEngineArea.NEWS_GEOPOLITICAL_EVENTS,[(metric,1)],[])
850 if normalized is not None and state not in {'READY_WITH_EVENTS','READY_NO_EVENTS'}:
851 # Stale discovery reduces coverage/confidence, but does not
852 # expire a still-relevant, independently persisted event.
853 result=result.model_copy(update={'status':AreaScoreStatus.PARTIAL})
854 return result
855 events: list[ResearchEvent] = []
856 seen: set[tuple[str, str, str]] = set()
857 for event in value.events:
858 event_at = event.event_date or event.published_at
859 if event.status == ResearchLifecycleStatus.REJECTED or event_at is None:
860 continue
861 age = value.evaluated_at - _aware(event_at)
862 if age < timedelta(0) or age > timedelta(days=CURRENT_NEWS_WINDOW_DAYS):
863 continue
864 if not _issuer_or_proven_exposure(event, value):
865 continue
866 key = (event.source_url.casefold(), event.title.strip().casefold(), event_at.date().isoformat())
867 if key in seen:
868 continue
869 seen.add(key)
870 events.append(event)
871 metrics = [(_event_metric(event, "CURRENT_EVENT_IMPACT_V1"), 1) for event in events]
872 return self._finish(value, RuleEngineArea.NEWS_GEOPOLITICAL_EVENTS, metrics, [] if metrics else ["RELEVANT_CURRENT_EVENT_WITHIN_30_DAYS"])
873
874 def _shareholding(self, value: StockRuleEngineInput) -> AreaScoreResult:
875 if not _is_india(value.profile):
876 return self._finish(value, RuleEngineArea.SHAREHOLDING, [], [], applicable=False)
877 if not value.shareholding:
878 return self._finish(value, RuleEngineArea.SHAREHOLDING, [], ["LATEST_VALID_SHAREHOLDING_PERIOD"])
879 snapshots = sorted(value.shareholding, key=lambda item: item.period_end)
880 latest = snapshots[-1]
881 latest_values = {entry.category: entry for entry in latest.values}
882 metrics: list[tuple[RuleMetricResult, int]] = []
883 promoter = latest_values.get(ShareholdingCategory.PROMOTER)
884 if promoter:
885 datum = _shareholding_datum(latest, promoter.percentage, _enum_text(promoter.category))
886 metrics.append((_metric_result("PROMOTER_HOLDING", datum, _higher_better(datum.value, ((0, 25), (20, 45), (40, 65), (55, 78), (70, 82))), "PROMOTER_HOLDING_LEVEL_V1", "PERCENT"), 25))
887 pledge = latest_values.get(ShareholdingCategory.PROMOTER_PLEDGE)
888 if pledge:
889 datum = _shareholding_datum(latest, pledge.percentage, _enum_text(pledge.category))
890 metrics.append((_metric_result("PROMOTER_PLEDGE", datum, _lower_better(datum.value, ((0, 95), (5, 75), (15, 48), (30, 20), (50, 5))), "PROMOTER_PLEDGE_V1", "PERCENT"), 35))
891 if len(snapshots) >= 2:
892 previous_values = {entry.category: entry for entry in snapshots[-2].values}
893 for category, name, weight in (
894 (ShareholdingCategory.PROMOTER, "PROMOTER_TREND", 20),
895 (ShareholdingCategory.FII_FPI, "FII_FPI_TREND", 10),
896 (ShareholdingCategory.DII, "DII_TREND", 10),
897 ):
898 current = latest_values.get(category)
899 previous = previous_values.get(category)
900 if current and previous:
901 change = current.percentage - previous.percentage
902 datum = _shareholding_datum(latest, change, name, extra_snapshot=snapshots[-2])
903 metrics.append((_metric_result(name, datum, _higher_better(change, ((-5, 15), (-2, 35), (0, 55), (2, 75), (5, 92))), f"{name}_QUARTERLY_V1", "PERCENTAGE_POINTS"), weight))
904 return self._finish(value, RuleEngineArea.SHAREHOLDING, metrics, [] if metrics else ["STRUCTURED_OWNERSHIP_VALUES"])
905
906 def _governance(self, value: StockRuleEngineInput) -> AreaScoreResult:
907 events = [event for event in value.events if _governance_event(event)]
908 metrics: list[tuple[RuleMetricResult, int]] = []
909 for event in events:
910 text = f"{event.title} {event.summary}".casefold()
911 resolved = any(_contains_term(text, term) for term in _RESOLUTION_TERMS) or event.status == ResearchLifecycleStatus.REJECTED
912 if resolved:
913 score = Decimal("60") if event.status != ResearchLifecycleStatus.REJECTED else Decimal("65")
914 else:
915 score = Decimal(str(_event_metric_score(event)))
916 datum = _event_datum(event)
917 metrics.append((_metric_result("GOVERNANCE_EVIDENCE", datum, float(score), "UNRESOLVED_GOVERNANCE_HISTORY_V1" if not resolved else "RESOLVED_GOVERNANCE_EVIDENCE_V1", "EVENT"), 1))
918 latest = value.shareholding[-1] if value.shareholding else None
919 if latest:
920 pledge = next((entry for entry in latest.values if entry.category == ShareholdingCategory.PROMOTER_PLEDGE), None)
921 if pledge:
922 datum = _shareholding_datum(latest, pledge.percentage, "PROMOTER_PLEDGE_GOVERNANCE")
923 metrics.append((_metric_result("PROMOTER_PLEDGE_GOVERNANCE", datum, _lower_better(datum.value, ((0, 95), (5, 75), (15, 45), (30, 18), (50, 5))), "PROMOTER_PLEDGE_GOVERNANCE_V1", "PERCENT"), 1))
924 return self._finish(value, RuleEngineArea.MANAGEMENT_GOVERNANCE, metrics, [] if metrics else ["STRUCTURED_GOVERNANCE_EVIDENCE"])
925
926 def _sector_macro(self, value: StockRuleEngineInput) -> AreaScoreResult:
927 metrics: list[tuple[RuleMetricResult, int]] = []
928 sector = _sector(value)
929 if sector:
930 datum = _Datum(Decimal("50"), "CANONICAL_IDENTITY", None, value.evaluated_at, f"canonical-sector:{value.profile.instrument_id}:{sector.casefold()}", "CATEGORY", 4)
931 metrics.append((_metric_result("CANONICAL_SECTOR_BASIS", datum, 50, "CANONICAL_SECTOR_NEUTRAL_BASIS_V1", "CATEGORY", display_value=sector), 20))
932 structured = _structured_data(value)
933 performance = _pick(structured, "sectorperformance", "sectorreturn")
934 if performance:
935 pct = _as_percent(performance.value)
936 metrics.append((_metric_result("SECTOR_PERFORMANCE", _derived_datum(pct, "PERCENT", performance, "normalized:sector-performance"), _trend_score(pct), "DURABLE_SECTOR_PERFORMANCE_V1", "PERCENT"), 45))
937 macro_events = [event for event in value.events if _is_macro_event(event) and _proven_macro_exposure(event, value)]
938 metrics.extend((_event_metric(event, "PROVEN_MACRO_EXPOSURE_IMPACT_V1"), 35) for event in macro_events)
939 return self._finish(value, RuleEngineArea.SECTOR_MACRO, metrics, [] if metrics else ["CANONICAL_SECTOR_OR_DURABLE_MACRO_EXPOSURE"])
940
941 def _finish(
942 self,
943 value: StockRuleEngineInput,
944 area: RuleEngineArea,
945 weighted_metrics: Sequence[tuple[RuleMetricResult, int]],
946 missing: Sequence[str],
947 *,
948 applicable: bool = True,
949 ) -> AreaScoreResult:
950 readiness = [item for item in value.readiness.requirements if item.rule_engine_area == area]
951 source_references = _readiness_source_references(readiness)
952 readiness_evidence = {
953 evidence_id for item in readiness for evidence_id in item.evidence_ids
954 }
955 readiness_missing = [entry for item in readiness for entry in item.missing_input_ids]
956 missing_inputs = sorted(set(missing) | set(readiness_missing))
957 if readiness and all(item.status == ResearchRequirementStatus.NOT_APPLICABLE for item in readiness):
958 return AreaScoreResult(area=area, weight=STOCK_RULE_ENGINE_AREA_WEIGHTS[area],
959 status=AreaScoreStatus.NOT_APPLICABLE, applicable=False,
960 evidence_references=[], source_references=[])
961 if not applicable or (readiness and all(item.status == ResearchRequirementStatus.UNSUPPORTED for item in readiness)):
962 return AreaScoreResult(
963 area=area,
964 weight=STOCK_RULE_ENGINE_AREA_WEIGHTS[area],
965 status=AreaScoreStatus.UNSUPPORTED,
966 applicable=False,
967 evidence_references=sorted(readiness_evidence),
968 source_references=source_references,
969 )
970 total_weight = sum(weight for _, weight in weighted_metrics)
971 metrics = [
972 item.model_copy(
973 update={
974 "configured_subrule_weight": weight,
975 "applied_weight_pct": _round_score(
976 Decimal(weight) * 100 / Decimal(total_weight)
977 ),
978 }
979 )
980 for item, weight in weighted_metrics
981 ]
982 if not metrics:
983 return AreaScoreResult(
984 area=area,
985 weight=STOCK_RULE_ENGINE_AREA_WEIGHTS[area],
986 status=AreaScoreStatus.UNSCORABLE,
987 applicable=True,
988 evidence_references=sorted(readiness_evidence),
989 source_references=source_references,
990 missing_inputs=missing_inputs,
991 )
992 raw = _round_score(sum(Decimal(str(item.score)) * weight for item, weight in weighted_metrics) / Decimal(total_weight))
993 if any(item.status == ResearchRequirementStatus.CONFLICTING for item in readiness):
994 status = AreaScoreStatus.CONFLICTING
995 elif any(item.status in {ResearchRequirementStatus.MISSING, ResearchRequirementStatus.PARTIAL, ResearchRequirementStatus.FAILED, ResearchRequirementStatus.REFRESHING} for item in readiness) or missing:
996 status = AreaScoreStatus.PARTIAL
997 elif any(item.status == ResearchRequirementStatus.READY_STALE for item in readiness):
998 status = AreaScoreStatus.READY_STALE
999 else:
1000 status = AreaScoreStatus.READY_FRESH
1001 evidence = sorted(
1002 readiness_evidence
1003 | {ref for item in metrics for ref in item.evidence_references}
1004 )
1005 positives = [f"{item.metric}: {item.score:.2f}/100" for item in metrics if item.score >= 65]
1006 negatives = [f"{item.metric}: {item.score:.2f}/100" for item in metrics if item.score <= 40]
1007 return AreaScoreResult(
1008 area=area,
1009 weight=STOCK_RULE_ENGINE_AREA_WEIGHTS[area],
1010 raw_score=raw,
1011 weighted_contribution=_round_score(raw * STOCK_RULE_ENGINE_AREA_WEIGHTS[area] / 100),
1012 status=status,
1013 applicable=True,
1014 metrics=metrics,
1015 positive_factors=positives,
1016 negative_factors=negatives,
1017 evidence_references=evidence,
1018 source_references=source_references,
1019 missing_inputs=missing_inputs,
1020 )
1021
1022 @staticmethod
1023 def _confidence(readiness: ResearchReadinessResult) -> float:
1024 supported = [item for item in readiness.requirements if item.status not in {ResearchRequirementStatus.UNSUPPORTED, ResearchRequirementStatus.NOT_APPLICABLE}]
1025 authority_points = {
1026 ResearchSourceTier.OFFICIAL: 100,
1027 ResearchSourceTier.REGULATORY: 100,
1028 ResearchSourceTier.TRUSTED_MARKET_DATA: 90,
1029 ResearchSourceTier.LICENSED_STRUCTURED: 80,
1030 ResearchSourceTier.APPROVED_SECONDARY: 65,
1031 ResearchSourceTier.APPROVED_EXTERNAL_TOOL: 50,
1032 ResearchSourceTier.USER_UPLOAD: 40,
1033 ResearchSourceTier.UNVERIFIED: 20,
1034 }
1035 freshness_points = {
1036 ResearchRequirementStatus.READY_FRESH: 100,
1037 ResearchRequirementStatus.READY_STALE: 55,
1038 ResearchRequirementStatus.PARTIAL: 40,
1039 ResearchRequirementStatus.CONFLICTING: 20,
1040 ResearchRequirementStatus.REFRESHING: 20,
1041 ResearchRequirementStatus.MISSING: 0,
1042 ResearchRequirementStatus.FAILED: 0,
1043 }
1044 sourced = [authority_points[item.source_tier] for item in supported if item.source_tier]
1045 authority = sum(sourced) / len(sourced) if sourced else 0
1046 freshness = sum(freshness_points.get(item.status, 0) for item in supported) / len(supported) if supported else 0
1047 conflicts = sum(item.status == ResearchRequirementStatus.CONFLICTING for item in supported)
1048 score = (
1049 Decimal(readiness.critical_completeness_pct) * Decimal("0.40")
1050 + Decimal(readiness.overall_completeness_pct) * Decimal("0.20")
1051 + Decimal(str(authority)) * Decimal("0.20")
1052 + Decimal(str(freshness)) * Decimal("0.20")
1053 - Decimal(min(30, conflicts * 15))
1054 )
1055 return _round_score(_clamp(score))
1056
1057 @staticmethod
1058 def _decision(
1059 score: float | None,
1060 readiness: ResearchReadinessResult,
1061 confidence: ConfidenceLevel,
1062 *,
1063 partial: bool,
1064 overrides: Sequence[RiskOverrideResult],
1065 ) -> DecisionSignal:
1066 if score is None:
1067 return DecisionSignal.INSUFFICIENT_DATA
1068 if any(item.effect == "BLOCK_BUY" for item in overrides):
1069 if any(item.severity == RiskOverrideSeverity.CRITICAL for item in overrides):
1070 return DecisionSignal.EXIT_REVIEW
1071 return DecisionSignal.AVOID
1072 signal = (
1073 DecisionSignal.STRONG_BUY if score >= 85 else
1074 DecisionSignal.BUY if score >= 75 else
1075 DecisionSignal.ACCUMULATE if score >= 65 else
1076 DecisionSignal.HOLD if score >= 50 else
1077 DecisionSignal.REDUCE if score >= 35 else
1078 DecisionSignal.AVOID if score >= 20 else
1079 DecisionSignal.EXIT_REVIEW
1080 )
1081 critical_fresh = all(
1082 readiness.for_requirement(key).status == ResearchRequirementStatus.READY_FRESH
1083 for key in StockRuleEngineEligibilityPolicy.CRITICAL_REQUIREMENTS
1084 )
1085 if signal == DecisionSignal.STRONG_BUY and (
1086 readiness.critical_completeness_pct < 100
1087 or confidence != ConfidenceLevel.HIGH
1088 or not critical_fresh
1089 ):
1090 signal = DecisionSignal.BUY
1091 if any(item.status == ResearchRequirementStatus.CONFLICTING and item.mandatory for item in readiness.requirements) and signal in {DecisionSignal.STRONG_BUY, DecisionSignal.BUY}:
1092 signal = DecisionSignal.HOLD
1093 if partial and signal in {DecisionSignal.STRONG_BUY, DecisionSignal.BUY, DecisionSignal.ACCUMULATE}:
1094 signal = DecisionSignal.HOLD
1095 return signal
1096
1097 def _risk_overrides(self, value: StockRuleEngineInput) -> list[RiskOverrideResult]:
1098 overrides: list[RiskOverrideResult] = []
1099 from app.news_intelligence import impact_at, latest_known_features
1100 for feature in latest_known_features(value.news_features,value.evaluated_at):
1101 if feature.severe_validated and impact_at(feature,value.evaluated_at) is not None:
1102 overrides.append(RiskOverrideResult(code='VALIDATED_'+feature.event_type,severity=RiskOverrideSeverity.CRITICAL,
1103 evidence_ids=[str(feature.feature_id)]))
1104 authoritative = [event for event in value.events if _authoritative_unresolved_event(event)]
1105 for event in authoritative:
1106 text = f"{event.title} {event.summary}".casefold()
1107 evidence = [f"event:{event.event_id}:GOVERNANCE_HISTORY"]
1108 if any(term in text for term in _FRAUD_TERMS) and event.impact == EventImpact.STRONG_NEGATIVE:
1109 overrides.append(RiskOverrideResult(code="CONFIRMED_FRAUD_OR_ACCOUNTING_CRISIS", severity=RiskOverrideSeverity.CRITICAL, evidence_ids=evidence))
1110 elif event.event_type == ResearchEventType.REGULATORY_EVENT and event.impact == EventImpact.STRONG_NEGATIVE and any(term in text for term in _CRITICAL_REGULATORY_TERMS):
1111 overrides.append(RiskOverrideResult(code="CRITICAL_REGULATORY_ACTION", severity=RiskOverrideSeverity.CRITICAL, evidence_ids=evidence))
1112 elif event.event_type == ResearchEventType.MANAGEMENT_CHANGE and event.impact == EventImpact.STRONG_NEGATIVE and any(term in text for term in ("auditor", "resignation", "qualification")):
1113 overrides.append(RiskOverrideResult(code="SEVERE_UNRESOLVED_GOVERNANCE", severity=RiskOverrideSeverity.HIGH, evidence_ids=evidence))
1114 if not _is_financial(value):
1115 extreme = _extreme_balance_sheet_evidence(value)
1116 if extreme:
1117 overrides.append(RiskOverrideResult(code="EXTREME_BALANCE_SHEET_STRESS", severity=RiskOverrideSeverity.CRITICAL, evidence_ids=extreme))
1118 negative_guidance = [event for event in authoritative if event.event_type in {ResearchEventType.GUIDANCE_CUT, ResearchEventType.ORDER_CANCELLED, ResearchEventType.PROJECT_DELAY}]
1119 independent = {event.independence_key or event.source_url for event in negative_guidance}
1120 if len(independent) >= 2:
1121 overrides.append(RiskOverrideResult(code="BROKEN_INVESTMENT_THESIS", severity=RiskOverrideSeverity.HIGH, evidence_ids=sorted(f"event:{event.event_id}:GOVERNANCE_HISTORY" for event in negative_guidance)))
1122 unique: dict[str, RiskOverrideResult] = {}
1123 for item in overrides:
1124 unique.setdefault(item.code, item)
1125 return list(unique.values())
1126
1127
1128 def analysis_eligibility_response(readiness: ResearchReadinessResult) -> dict[str, Any]:
1129 return StockRuleEngineEligibilityPolicy().evaluate(readiness).model_dump(
1130 mode="json", by_alias=True
1131 )
1132
1133
1134 def _readiness_source_references(
1135 readiness: Sequence[ResearchRequirementReadiness],
1136 ) -> list[RuleSourceReference]:
1137 grouped: dict[
1138 tuple[str | None, ResearchSourceTier | None, str, datetime | None, datetime | None],
1139 set[str],
1140 ] = {}
1141 for item in readiness:
1142 if not item.source_url:
1143 continue
1144 key = (
1145 item.source,
1146 item.source_tier,
1147 item.source_url,
1148 item.as_of,
1149 item.retrieved_at,
1150 )
1151 grouped.setdefault(key, set()).update(item.evidence_ids)
1152 return [
1153 RuleSourceReference(
1154 source_provider=source,
1155 source_tier=source_tier,
1156 source_url=source_url,
1157 as_of=as_of,
1158 retrieved_at=retrieved_at,
1159 evidence_references=sorted(evidence_ids),
1160 )
1161 for (source, source_tier, source_url, as_of, retrieved_at), evidence_ids in sorted(
1162 grouped.items(), key=lambda entry: (
1163 entry[0][0] or "",
1164 entry[0][2],
1165 _iso(entry[0][3]) or "",
1166 _iso(entry[0][4]) or "",
1167 )
1168 )
1169 ]
1170
1171
1172 def _structured_data(value: StockRuleEngineInput) -> dict[str, _Datum]:
1173 selected: dict[str, _Datum] = {}
1174 for record in value.structured_snapshots:
1175 authority = _provider_authority(record.provider, record.source_type, record.source_name)
1176 for raw_key, fact in record.snapshot.facts.items():
1177 number = _decimal(fact.value)
1178 if number is None:
1179 continue
1180 if raw_key == "latestPrice" and number <= 0:
1181 continue
1182 key = _metric_key(raw_key)
1183 datum = _Datum(
1184 number,
1185 fact.source_name or record.source_name or record.provider,
1186 fact.source_url or record.source_url,
1187 fact.as_of_date or record.market_as_of,
1188 f"structured:{record.provider}:{raw_key}:{record.retrieved_at.isoformat()}",
1189 fact.unit,
1190 authority,
1191 )
1192 existing = selected.get(key)
1193 if existing is None or (datum.authority, datum.as_of or datetime.min.replace(tzinfo=timezone.utc), datum.evidence_id) > (existing.authority, existing.as_of or datetime.min.replace(tzinfo=timezone.utc), existing.evidence_id):
1194 selected[key] = datum
1195 return selected
1196
1197
1198 def _pick(values: Mapping[str, _Datum], *keys: str) -> _Datum | None:
1199 return next((values[_metric_key(key)] for key in keys if _metric_key(key) in values), None)
1200
1201
1202 def _latest_fact(
1203 value: StockRuleEngineInput,
1204 metrics: Sequence[str],
1205 *,
1206 period_type: str | None = None,
1207 ) -> _Datum | None:
1208 series = _fact_series(value, metrics, period_type)
1209 return series[-1] if series else None
1210
1211
1212 def _fact_series(
1213 value: StockRuleEngineInput,
1214 metrics: Sequence[str],
1215 period_type: str | None,
1216 ) -> list[_Datum]:
1217 keys = {_metric_key(metric) for metric in metrics}
1218 selected: dict[str, tuple[int, _Datum]] = {}
1219 for fact in value.financial_facts:
1220 if fact.source_mode != SourceMode.REAL or _metric_key(fact.key.metric) not in keys:
1221 continue
1222 if period_type and fact.key.period_type.upper() != period_type.upper():
1223 continue
1224 number = _decimal(fact.value.value)
1225 as_of = fact.value.as_of_date or _period_datetime(fact.key.period_end)
1226 if number is None or as_of is None:
1227 continue
1228 datum = _Datum(
1229 number,
1230 fact.source_provider,
1231 fact.value.source_url,
1232 as_of,
1233 f"financial:{fact.source_identity}:{fact.key.metric}:{fact.key.period_end}:{fact.key.period_type}",
1234 fact.value.unit,
1235 fact_source_authority(fact.source_tier),
1236 )
1237 period_key = f"{as_of.isoformat()}:{fact.key.reporting_basis or ''}"
1238 existing = selected.get(period_key)
1239 if existing is None or datum.authority > existing[0]:
1240 selected[period_key] = (datum.authority, datum)
1241 return sorted((entry[1] for entry in selected.values()), key=lambda item: item.as_of or datetime.min.replace(tzinfo=timezone.utc))
1242
1243
1244 def _metric_result(
1245 name: str,
1246 datum: _Datum,
1247 score: float | Decimal,
1248 rule: str,
1249 unit: str | None,
1250 *,
1251 display_value: Any | None = None,
1252 ) -> RuleMetricResult:
1253 return RuleMetricResult(
1254 metric=name,
1255 value=_json_number(datum.value) if display_value is None else display_value,
1256 unit=unit or datum.unit,
1257 score=_round_score(score),
1258 rule=rule,
1259 source=datum.source,
1260 source_url=datum.source_url,
1261 as_of=datum.as_of,
1262 evidence_references=sorted(set(datum.evidence_id.split("|"))),
1263 )
1264
1265
1266 def _derived_datum(value: Decimal, unit: str, primary: _Datum, evidence_id: str, *others: _Datum) -> _Datum:
1267 values = (primary, *others)
1268 return _Datum(
1269 value,
1270 primary.source,
1271 primary.source_url or next((item.source_url for item in others if item.source_url), None),
1272 max((item.as_of for item in values if item.as_of), default=None),
1273 "|".join([evidence_id, *(item.evidence_id for item in values)]),
1274 unit,
1275 max(item.authority for item in values),
1276 )
1277
1278
1279 def _derived_from_many(value: Decimal, unit: str, inputs: Sequence[_Datum], evidence_id: str) -> _Datum:
1280 primary = inputs[-1]
1281 return _derived_datum(value, unit, primary, evidence_id, *inputs[:-1])
1282
1283
1284 def _price_datum(value: MarketPriceObservation) -> _Datum:
1285 return _Datum(value.price, value.provider, value.source_url, _aware(value.observed_at), f"price:{value.provider}:{value.observed_at.isoformat()}", value.currency, _provider_authority(value.provider, None, None))
1286
1287
1288 def _shareholding_datum(snapshot: ShareholdingSnapshot, number: Decimal, label: str, *, extra_snapshot: ShareholdingSnapshot | None = None) -> _Datum:
1289 ids = [f"shareholding:{snapshot.id}"]
1290 if extra_snapshot:
1291 ids.append(f"shareholding:{extra_snapshot.id}")
1292 return _Datum(number, snapshot.source_provider, snapshot.source_url, snapshot.period_end, "|".join(ids), "PERCENT", 5 if snapshot.source_provider.upper() == "NSE" else 3)
1293
1294
1295 def _event_datum(event: ResearchEvent) -> _Datum:
1296 event_at = event.event_date or event.published_at or event.detected_at
1297 return _Datum(Decimal(str(_event_metric_score(event))), _enum_text(event.source_classification), event.source_url, _aware(event_at), f"event:{event.event_id}", "EVENT", _event_authority(event))
1298
1299
1300 def _event_metric(event: ResearchEvent, rule: str) -> RuleMetricResult:
1301 datum = _event_datum(event)
1302 return _metric_result(_enum_text(event.event_type), datum, _event_metric_score(event), rule, "EVENT", display_value={"direction": _enum_text(event.impact), "confidence": event.confidence})
1303
1304
1305 def _event_metric_score(event: ResearchEvent) -> float:
1306 direction = {
1307 EventImpact.STRONG_POSITIVE: Decimal("1"),
1308 EventImpact.POSITIVE: Decimal("0.6"),
1309 EventImpact.NEUTRAL: Decimal("0"),
1310 EventImpact.UNCERTAIN: Decimal("0"),
1311 EventImpact.NEGATIVE: Decimal("-0.6"),
1312 EventImpact.STRONG_NEGATIVE: Decimal("-1"),
1313 }[event.impact]
1314 severity = Decimal("1") if event.event_type in {
1315 ResearchEventType.MAJOR_CONTRACT,
1316 ResearchEventType.GOVERNMENT_CONTRACT,
1317 ResearchEventType.ORDER_CANCELLED,
1318 ResearchEventType.PROJECT_DELAY,
1319 ResearchEventType.GUIDANCE_RAISED,
1320 ResearchEventType.GUIDANCE_CUT,
1321 ResearchEventType.REGULATORY_EVENT,
1322 ResearchEventType.MANAGEMENT_CHANGE,
1323 } else Decimal("0.75")
1324 if event.monetary_value is not None or event.capacity_value is not None or event.percentage_value is not None:
1325 severity = min(Decimal("1"), severity + Decimal("0.1"))
1326 reliability = {
1327 ReliabilityLevel.LEVEL_A: Decimal("1"),
1328 ReliabilityLevel.LEVEL_B: Decimal("0.85"),
1329 ReliabilityLevel.LEVEL_C: Decimal("0.65"),
1330 ReliabilityLevel.LEVEL_D: Decimal("0.4"),
1331 ReliabilityLevel.LEVEL_E: Decimal("0.2"),
1332 }[event.reliability]
1333 duration = {
1334 TimeHorizon.IMMEDIATE: Decimal("0.75"),
1335 TimeHorizon.SHORT_TERM: Decimal("0.85"),
1336 TimeHorizon.MEDIUM_TERM: Decimal("1"),
1337 TimeHorizon.LONG_TERM: Decimal("1"),
1338 TimeHorizon.UNKNOWN: Decimal("0.6"),
1339 }[event.time_horizon]
1340 probability = Decimal(str(event.confidence)) * reliability
1341 # V1 has no durable priced-in field. It applies a documented neutral factor
1342 # of 1.0 instead of guessing one.
1343 return _round_score(_clamp(Decimal("50") + direction * Decimal("50") * severity * probability * duration))
1344
1345
1346 def _material_catalyst(event: ResearchEvent) -> bool:
1347 if event.source_mode != SourceMode.REAL or event.status != ResearchLifecycleStatus.VALIDATED or event.confidence < 0.6:
1348 return False
1349 if event.reliability not in {ReliabilityLevel.LEVEL_A, ReliabilityLevel.LEVEL_B}:
1350 return False
1351 if event.event_type not in _CATALYST_TYPES:
1352 return False
1353 explicit_materiality = any((event.monetary_value is not None, event.capacity_value is not None, event.percentage_value is not None, bool(event.customer), bool(event.counterparty)))
1354 authoritative = event.source_classification in _AUTHORITATIVE_CLASSIFICATIONS
1355 return explicit_materiality or authoritative
1356
1357
1358 def _issuer_or_proven_exposure(event: ResearchEvent, value: StockRuleEngineInput) -> bool:
1359 if event.source_mode != SourceMode.REAL or event.confidence < 0.5:
1360 return False
1361 return _proven_macro_exposure(event, value) if _is_macro_event(event) else True
1362
1363
1364 def _is_macro_event(event: ResearchEvent) -> bool:
1365 text = f"{event.title} {event.summary}".casefold()
1366 return any(_contains_term(text, term) for term in _MACRO_TERMS)
1367
1368
1369 def _proven_macro_exposure(event: ResearchEvent, value: StockRuleEngineInput) -> bool:
1370 text = f"{event.title} {event.summary} {event.location or ''} {event.counterparty or ''}".casefold()
1371 sector = (_sector(value) or "").casefold()
1372 exposures = {
1373 "energy": ("oil", "crude", "gas", "commodity"),
1374 "airline": ("oil", "fuel", "jet fuel"),
1375 "aviation": ("oil", "fuel", "jet fuel"),
1376 "financial": ("interest rate", "central bank", "credit"),
1377 "bank": ("interest rate", "central bank", "credit"),
1378 "materials": ("commodity", "tariff", "supply chain"),
1379 "industrial": ("tariff", "supply chain", "commodity"),
1380 "technology": ("sanction", "tariff", "supply chain", "currency"),
1381 "consumer": ("currency", "commodity", "interest rate"),
1382 "automobile": ("commodity", "tariff", "supply chain"),
1383 }
1384 allowed = {term for label, terms in exposures.items() if label in sector for term in terms}
1385 # A durable structured relationship is required: both a recognized sector
1386 # sensitivity and that same sensitivity in the event evidence.
1387 return bool(allowed and any(_contains_term(text, term) for term in allowed))
1388
1389
1390 def _governance_event(event: ResearchEvent) -> bool:
1391 text = f"{event.title} {event.summary}".casefold()
1392 return event.event_type in {ResearchEventType.MANAGEMENT_CHANGE, ResearchEventType.REGULATORY_EVENT, ResearchEventType.CREDIT_RATING, ResearchEventType.GUIDANCE_CUT} or any(term in text for term in _GOVERNANCE_TERMS)
1393
1394
1395 def _authoritative_unresolved_event(event: ResearchEvent) -> bool:
1396 text = f"{event.title} {event.summary}".casefold()
1397 return (
1398 event.source_mode == SourceMode.REAL
1399 and event.status == ResearchLifecycleStatus.VALIDATED
1400 and event.reliability == ReliabilityLevel.LEVEL_A
1401 and event.source_classification in _AUTHORITATIVE_CLASSIFICATIONS
1402 and event.confidence >= 0.75
1403 and not any(_contains_term(text, term) for term in _RESOLUTION_TERMS)
1404 )
1405
1406
1407 def _extreme_balance_sheet_evidence(value: StockRuleEngineInput) -> list[str]:
1408 structured = _structured_data(value)
1409 de = _pick(structured, "debttoequity")
1410 if de:
1411 de_pct = de.value if abs(de.value) > 5 else de.value * 100
1412 else:
1413 debt = _latest_fact(value, ("debt_or_borrowings", "total_debt"), period_type="ANNUAL")
1414 equity = _latest_fact(value, ("equity", "total_equity"), period_type="ANNUAL")
1415 if not debt or not equity or equity.value <= 0:
1416 return []
1417 de_pct = debt.value / equity.value * 100
1418 de = _derived_datum(de_pct, "PERCENT", debt, "derived:override-debt-equity", equity)
1419 ebitda = _latest_fact(value, ("ebitda", "ebit", "operating_income"), period_type="ANNUAL")
1420 finance_cost = _latest_fact(value, ("finance_cost", "interest_expense"), period_type="ANNUAL")
1421 if de_pct <= 300 or not ebitda or not finance_cost or finance_cost.value <= 0 or ebitda.value / finance_cost.value >= 1:
1422 return []
1423 # Both facts must originate from an official/regulatory durable tier.
1424 if de.authority < fact_source_authority(FactSourceTier.OFFICIAL_NSE) or ebitda.authority < fact_source_authority(FactSourceTier.OFFICIAL_NSE) or finance_cost.authority < fact_source_authority(FactSourceTier.OFFICIAL_NSE):
1425 return []
1426 return sorted(set(de.evidence_id.split("|") + ebitda.evidence_id.split("|") + finance_cost.evidence_id.split("|")))
1427
1428
1429 def _area_weighted_score(values: Sequence[AreaScoreResult], *, areas: set[RuleEngineArea] | None = None) -> float | None:
1430 included = [item for item in values if item.raw_score is not None and (areas is None or item.area in areas)]
1431 total = sum(item.weight for item in included)
1432 if not total:
1433 return None
1434 return _round_score(sum(Decimal(str(item.raw_score)) * item.weight for item in included) / Decimal(total))
1435
1436
1437 def _custom_area_score(values: Sequence[AreaScoreResult], weights: Mapping[RuleEngineArea, int]) -> float | None:
1438 included = [item for item in values if item.raw_score is not None and item.area in weights]
1439 total = sum(weights[item.area] for item in included)
1440 if not total:
1441 return None
1442 return _round_score(sum(Decimal(str(item.raw_score)) * weights[item.area] for item in included) / Decimal(total))
1443
1444
1445 def _comparable_yoy(series: Sequence[_Datum]) -> tuple[Decimal, _Datum, _Datum] | None:
1446 if len(series) < 2:
1447 return None
1448 latest = series[-1]
1449 candidates = [item for item in series[:-1] if item.as_of and latest.as_of and 300 <= (latest.as_of - item.as_of).days <= 430]
1450 if not candidates:
1451 return None
1452 prior = min(candidates, key=lambda item: abs((latest.as_of - item.as_of).days - 365))
1453 if prior.value == 0:
1454 return None
1455 return (latest.value / abs(prior.value) - 1) * 100, latest, prior
1456
1457
1458 def _sequential_change(series: Sequence[_Datum]) -> tuple[Decimal, _Datum, _Datum] | None:
1459 if len(series) < 2 or series[-2].value == 0:
1460 return None
1461 return (series[-1].value / abs(series[-2].value) - 1) * 100, series[-1], series[-2]
1462
1463
1464 def _series_cagr(series: Sequence[_Datum]) -> Decimal | None:
1465 if len(series) < 3 or series[0].value <= 0 or series[-1].value <= 0 or not series[0].as_of or not series[-1].as_of:
1466 return None
1467 years = Decimal(str((series[-1].as_of - series[0].as_of).days / 365.25))
1468 if years <= Decimal("1"):
1469 return None
1470 return (Decimal(str(math.pow(float(series[-1].value / series[0].value), 1 / float(years)))) - 1) * 100
1471
1472
1473 def _growth_consistency(*series_values: Sequence[_Datum]) -> tuple[Decimal, list[_Datum]] | None:
1474 changes: list[bool] = []
1475 evidence: list[_Datum] = []
1476 for series in series_values:
1477 for previous, current in zip(series[-5:-1], series[-4:]):
1478 changes.append(current.value >= previous.value)
1479 evidence.extend((previous, current))
1480 if not changes:
1481 return None
1482 return Decimal(sum(changes)) / Decimal(len(changes)) * 100, evidence
1483
1484
1485 def _aligned_ratios(numerators: Sequence[_Datum], denominators: Sequence[_Datum], *, multiplier: Decimal) -> list[Decimal]:
1486 denominator_by_date = {item.as_of.date(): item for item in denominators if item.as_of and item.value != 0}
1487 ratios: list[Decimal] = []
1488 for numerator in numerators:
1489 denominator = denominator_by_date.get(numerator.as_of.date()) if numerator.as_of else None
1490 if denominator:
1491 ratios.append(numerator.value / denominator.value * multiplier)
1492 return ratios
1493
1494
1495 def _usable_prices(values: Iterable[MarketPriceObservation]) -> list[MarketPriceObservation]:
1496 return sorted(
1497 (
1498 item for item in values
1499 if item.price is not None and Decimal(str(item.price)).is_finite() and item.price > 0
1500 ),
1501 key=lambda item: item.observed_at,
1502 )
1503
1504
1505 def _lower_better(value: Decimal, points: Sequence[tuple[float, float]]) -> float:
1506 return _piecewise(value, points)
1507
1508
1509 def _higher_better(value: Decimal, points: Sequence[tuple[float, float]]) -> float:
1510 return _piecewise(value, points)
1511
1512
1513 def _piecewise(value: Decimal, points: Sequence[tuple[float, float]]) -> float:
1514 ordered = [(Decimal(str(x)), Decimal(str(y))) for x, y in points]
1515 if value <= ordered[0][0]:
1516 return _round_score(ordered[0][1])
1517 if value >= ordered[-1][0]:
1518 return _round_score(ordered[-1][1])
1519 for (left_x, left_y), (right_x, right_y) in zip(ordered, ordered[1:]):
1520 if left_x <= value <= right_x:
1521 ratio = (value - left_x) / (right_x - left_x)
1522 return _round_score(left_y + ratio * (right_y - left_y))
1523 return 50.0
1524
1525
1526 def _growth_score(value: Decimal) -> float:
1527 return _higher_better(value, ((-30, 5), (-10, 25), (0, 45), (10, 65), (20, 80), (40, 95)))
1528
1529
1530 def _trend_score(value: Decimal) -> float:
1531 # Momentum gets no extra reward beyond +30%; V1 avoids rewarding parabolic
1532 # distance indefinitely.
1533 return _higher_better(value, ((-40, 8), (-20, 25), (-5, 45), (0, 55), (10, 72), (30, 88), (60, 78)))
1534
1535
1536 def _provider_authority(provider: str | None, source_type: str | None, source_name: str | None) -> int:
1537 text = f"{provider or ''} {source_type or ''} {source_name or ''}".upper()
1538 if "SEC" in text or "REGULATORY" in text:
1539 return 5
1540 if "NSE" in text or "EXCHANGE" in text or "COMPANY FILING" in text:
1541 return 4
1542 if "EODHD" in text or "LICENSED" in text:
1543 return 3
1544 if "YAHOO" in text:
1545 return 2
1546 return 1
1547
1548
1549 def _event_authority(event: ResearchEvent) -> int:
1550 if event.source_classification in {SourceClassification.REGULATORY, SourceClassification.EXCHANGE}:
1551 return 5
1552 if event.source_classification == SourceClassification.OFFICIAL_COMPANY:
1553 return 4
1554 if event.source_classification == SourceClassification.REPUTABLE_NEWS:
1555 return 2
1556 return 1
1557
1558
1559 def _sector(value: StockRuleEngineInput) -> str | None:
1560 metadata = value.canonical_metadata
1561 raw = metadata.get("canonicalSector") or metadata.get("sector") or metadata.get("industrySector")
1562 if raw:
1563 return str(raw).strip()
1564 for key in ("sector", "industry"):
1565 for record in value.structured_snapshots:
1566 item = next((fact for raw_key, fact in record.snapshot.facts.items() if _metric_key(raw_key) == key and fact.value), None)
1567 if item:
1568 return str(item.value).strip()
1569 return None
1570
1571
1572 def _is_financial(value: StockRuleEngineInput) -> bool:
1573 text = f"{_sector(value) or ''} {value.canonical_metadata.get('industry') or ''}".casefold()
1574 return any(token in text for token in ("bank", "financial", "nbfc", "insurance", "credit service"))
1575
1576
1577 def _is_india(profile: CompanyResearchProfile) -> bool:
1578 return profile.country.strip().upper() in {"IN", "IND", "INDIA"} or profile.exchange.strip().upper() in {"NSE", "XNSE", "BSE", "XBOM"}
1579
1580
1581 def _input_as_of(value: StockRuleEngineInput) -> datetime | None:
1582 dates: list[datetime] = []
1583 dates.extend(item.observed_at for item in _usable_prices(value.market_prices))
1584 dates.extend(item.value.as_of_date or _period_datetime(item.key.period_end) for item in value.financial_facts if item.value.as_of_date or item.key.period_end)
1585 dates.extend(item.market_as_of for item in value.structured_snapshots if item.market_as_of)
1586 dates.extend(item.event_date or item.published_at for item in value.events if item.event_date or item.published_at)
1587 dates.extend(item.period_end for item in value.shareholding)
1588 return max((_aware(item) for item in dates if item is not None), default=None)
1589
1590
1591 def _confidence_level(score: float) -> ConfidenceLevel:
1592 return ConfidenceLevel.HIGH if score >= 80 else ConfidenceLevel.MEDIUM if score >= 55 else ConfidenceLevel.LOW
1593
1594
1595 def _metric_key(value: str) -> str:
1596 return re.sub(r"[^a-z0-9]", "", str(value).casefold())
1597
1598
1599 def _enum_text(value: Any) -> str:
1600 return str(getattr(value, "value", value))
1601
1602
1603 def _contains_term(text: str, term: str) -> bool:
1604 return re.search(rf"(?<![a-z0-9]){re.escape(term.casefold())}(?![a-z0-9])", text) is not None
1605
1606
1607 def _decimal(value: Any) -> Decimal | None:
1608 try:
1609 number = Decimal(str(value))
1610 except (InvalidOperation, TypeError, ValueError):
1611 return None
1612 return number if number.is_finite() else None
1613
1614
1615 def _as_percent(value: Decimal) -> Decimal:
1616 return value * 100 if abs(value) <= Decimal("2") else value
1617
1618
1619 def _clamp(value: Decimal) -> Decimal:
1620 return max(Decimal("0"), min(Decimal("100"), value))
1621
1622
1623 def _round_score(value: float | Decimal) -> float:
1624 return float(Decimal(str(value)).quantize(Decimal("0.01")))
1625
1626
1627 def _json_number(value: Decimal) -> float:
1628 return float(value)
1629
1630
1631 def _aware(value: datetime) -> datetime:
1632 return value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value.astimezone(timezone.utc)
1633
1634
1635 def _period_datetime(value: str | None) -> datetime | None:
1636 if not value:
1637 return None
1638 try:
1639 parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
1640 except ValueError:
1641 try:
1642 parsed = datetime.strptime(str(value), "%Y-%m-%d")
1643 except ValueError:
1644 return None
1645 return _aware(parsed)
1646
1647
1648 def _iso(value: datetime | None) -> str | None:
1649 return _aware(value).isoformat() if value else None
1650
1651
1652 _AUTHORITATIVE_CLASSIFICATIONS = {
1653 SourceClassification.OFFICIAL_COMPANY,
1654 SourceClassification.REGULATORY,
1655 SourceClassification.EXCHANGE,
1656 }
1657 _CATALYST_TYPES = {
1658 ResearchEventType.ORDER_WIN,
1659 ResearchEventType.NEW_CONTRACT,
1660 ResearchEventType.CLIENT_WIN,
1661 ResearchEventType.NEW_PLANT,
1662 ResearchEventType.MAJOR_CORPORATE_ANNOUNCEMENT,
1663 ResearchEventType.NEW_ORDER,
1664 ResearchEventType.ORDER_BACKLOG_CHANGE,
1665 ResearchEventType.NEW_CUSTOMER,
1666 ResearchEventType.CUSTOMER_EXPANSION,
1667 ResearchEventType.MAJOR_CUSTOMER,
1668 ResearchEventType.CUSTOMER_LOSS,
1669 ResearchEventType.MAJOR_CONTRACT,
1670 ResearchEventType.GOVERNMENT_CONTRACT,
1671 ResearchEventType.CAPEX,
1672 ResearchEventType.FACTORY_EXPANSION,
1673 ResearchEventType.CAPACITY_EXPANSION,
1674 ResearchEventType.NEW_FACILITY,
1675 ResearchEventType.GEOGRAPHIC_EXPANSION,
1676 ResearchEventType.ACQUISITION,
1677 ResearchEventType.PARTNERSHIP,
1678 ResearchEventType.PRODUCT_LAUNCH,
1679 ResearchEventType.GUIDANCE_RAISED,
1680 ResearchEventType.GUIDANCE_LOWERED,
1681 ResearchEventType.GUIDANCE_MAINTAINED,
1682 ResearchEventType.GUIDANCE_CUT,
1683 ResearchEventType.ORDER_CANCELLED,
1684 ResearchEventType.PROJECT_DELAY,
1685 }
1686 _MACRO_TERMS = (
1687 "war", "sanction", "tariff", "interest rate", "central bank", "commodity",
1688 "oil", "crude", "currency", "supply chain", "geopolitical",
1689 )
1690 _GOVERNANCE_TERMS = (
1691 "fraud", "accounting", "auditor", "regulatory action", "enforcement",
1692 "litigation", "promoter pledge", "related party", "governance",
1693 )
1694 _FRAUD_TERMS = ("confirmed fraud", "accounting fraud", "accounting crisis", "financial misstatement")
1695 _CRITICAL_REGULATORY_TERMS = ("license revoked", "trading ban", "insolvency", "criminal enforcement", "operations suspended")
1696 _RESOLUTION_TERMS = ("resolved", "remediated", "cleared", "settled and closed", "no wrongdoing")