@prakhar-gitlab / ai-investment-platform
Code Commits Issues Pulls Sessions CI/CD
main
main
py 251 lines 15.2 KB
Raw
1 """Pure, date-aligned stock/sector/market relative returns.
2
3 Option B: missing durable benchmark mappings/history remain unavailable. A
4 sector performer leaderboard is not a sector index and is never substituted.
5 Comparisons are local-currency price returns (no invented FX conversion).
6 """
7 from __future__ import annotations
8
9 from dataclasses import asdict, dataclass
10 from datetime import date, datetime, timedelta
11 from math import isfinite
12 from typing import Iterable, Literal, Mapping
13 from uuid import UUID
14
15 from pydantic import Field
16
17 from app.models import DailyMarketBar, MarketPriceObservation, ResearchBaseModel
18 from app.sector_leaderboard import normalize_sector
19 from app.technical_features import normalize_price_history, normalize_daily_history, percentage, utc
20 from zoneinfo import ZoneInfo
21
22
23 SECTOR_FEATURE_VERSION = "SECTOR_RELATIVE_STRENGTH_V1"
24
25
26 @dataclass(frozen=True)
27 class BenchmarkReference:
28 instrument_id: UUID
29 currency: str
30 trusted_providers: frozenset[str] | None = None
31
32
33 @dataclass(frozen=True)
34 class SectorContext:
35 """Authoritative classification and explicit canonical mappings, supplied by caller.
36
37 India uses canonicalSector from the persisted NSE universe. Other markets
38 may use their persisted canonical classification. No company-name inference.
39 source/as_of retain classification provenance; future metadata is rejected.
40 """
41 sector: str | None = None
42 source: str | None = None
43 as_of: datetime | None = None
44 region: str | None = None
45 sector_benchmark: BenchmarkReference | None = None
46 market_benchmark: BenchmarkReference | None = None
47 mapping_version: str | None = None
48 sector_mapping_status: str | None = None
49 market_mapping_status: str | None = None
50
51
52 @dataclass(frozen=True)
53 class SectorRelativeStrengthConfig:
54 return_lookbacks: tuple[int, ...] = (5, 21, 63, 126)
55 period_weights: tuple[float, ...] = (1.0, 1.0, 1.0, 1.0)
56 max_age_days: int = 7
57 # Normalize horizon lengths before classifying direction: 0.02 percentage
58 # points/observed session is a configurable noise band, not a forecast.
59 material_edge_per_observation_pct: float = 0.02
60 improvement_per_observation_pct: float = 0.02
61 # An average 0.2pp/session edge reaches score100, -0.2 reaches zero.
62 full_scale_edge_per_observation_pct: float = 0.2
63
64 def __post_init__(self):
65 if (len(self.return_lookbacks) != 4 or any(n < 1 for n in self.return_lookbacks)
66 or tuple(sorted(set(self.return_lookbacks))) != self.return_lookbacks
67 or len(self.period_weights) != 4 or sum(self.period_weights) <= 0 or self.max_age_days < 1
68 or self.full_scale_edge_per_observation_pct <= 0):
69 raise ValueError("Invalid sector configuration")
70 for value in asdict(self).values():
71 if any(not isfinite(v) or v < 0 for v in (value if isinstance(value, tuple) else (value,))):
72 raise ValueError("Sector thresholds/weights must be finite and nonnegative")
73
74
75 class SectorRelativeStrengthSnapshot(ResearchBaseModel):
76 global_instrument_id: UUID
77 as_of: datetime
78 feature_version: str = SECTOR_FEATURE_VERSION
79 configuration: dict
80 benchmark_mapping_version: str | None = None
81 benchmark_states: dict[str, str] = Field(default_factory=dict)
82 history_sources: dict[str, str] = Field(default_factory=dict)
83 sector: str | None = None
84 classification_source: str | None = None
85 classification_as_of: datetime | None = None
86 region: str | None = None
87 sector_benchmark_id: UUID | None = None
88 market_benchmark_id: UUID | None = None
89 return_basis: str = "DATE_ALIGNED_LOCAL_CURRENCY_PRICE_RETURN_PCT"
90 stock_return1_w: float | None = Field(default=None, alias="stockReturn1W")
91 stock_return1_m: float | None = Field(default=None, alias="stockReturn1M")
92 stock_return3_m: float | None = Field(default=None, alias="stockReturn3M")
93 stock_return6_m: float | None = Field(default=None, alias="stockReturn6M")
94 sector_return1_w: float | None = Field(default=None, alias="sectorReturn1W")
95 sector_return1_m: float | None = Field(default=None, alias="sectorReturn1M")
96 sector_return3_m: float | None = Field(default=None, alias="sectorReturn3M")
97 sector_return6_m: float | None = Field(default=None, alias="sectorReturn6M")
98 market_return1_w: float | None = Field(default=None, alias="marketReturn1W")
99 market_return1_m: float | None = Field(default=None, alias="marketReturn1M")
100 market_return3_m: float | None = Field(default=None, alias="marketReturn3M")
101 market_return6_m: float | None = Field(default=None, alias="marketReturn6M")
102 relative_vs_sector1_w: float | None = Field(default=None, alias="relativeVsSector1W")
103 relative_vs_sector1_m: float | None = Field(default=None, alias="relativeVsSector1M")
104 relative_vs_sector3_m: float | None = Field(default=None, alias="relativeVsSector3M")
105 relative_vs_sector6_m: float | None = Field(default=None, alias="relativeVsSector6M")
106 relative_vs_market1_w: float | None = Field(default=None, alias="relativeVsMarket1W")
107 relative_vs_market1_m: float | None = Field(default=None, alias="relativeVsMarket1M")
108 relative_vs_market3_m: float | None = Field(default=None, alias="relativeVsMarket3M")
109 relative_vs_market6_m: float | None = Field(default=None, alias="relativeVsMarket6M")
110 comparison_windows: dict[str, tuple[date, date]] = Field(default_factory=dict)
111 feature_states: dict[str, str] = Field(default_factory=dict)
112 sector_state: Literal["LEADING", "IMPROVING", "NEUTRAL", "WEAKENING", "LAGGING", "INSUFFICIENT_DATA"] = "INSUFFICIENT_DATA"
113 relative_strength_score: float | None = None
114 confidence: float = 0
115 missing_inputs: list[str] = Field(default_factory=list)
116 stale_inputs: list[str] = Field(default_factory=list)
117
118
119 class SectorRelativeStrengthEngine:
120 def __init__(self, config: SectorRelativeStrengthConfig | None = None):
121 self.config = config or SectorRelativeStrengthConfig()
122
123 def compute(self, instrument_id: UUID, stock_history: Iterable[MarketPriceObservation], *, as_of: datetime,
124 context: SectorContext | None = None, currency: str | None = None,
125 benchmark_histories: Mapping[UUID, Iterable[MarketPriceObservation]] | None = None,
126 trusted_providers: frozenset[str] | None = None,
127 daily_bar_histories: Mapping[UUID, Iterable[DailyMarketBar]] | None = None) -> SectorRelativeStrengthSnapshot:
128 cfg, context = self.config, context or SectorContext()
129 histories = benchmark_histories or {}
130 daily = daily_bar_histories or {}
131 classification_valid = bool(context.sector and context.source and context.as_of is not None and utc(context.as_of) <= utc(as_of))
132 result = SectorRelativeStrengthSnapshot(global_instrument_id=instrument_id, as_of=utc(as_of), configuration=asdict(cfg),
133 sector=normalize_sector(context.sector)[1] if classification_valid else None,
134 classification_source=context.source if classification_valid else None,
135 classification_as_of=utc(context.as_of) if classification_valid else None, region=context.region,
136 sector_benchmark_id=context.sector_benchmark.instrument_id if context.sector_benchmark and classification_valid else None,
137 market_benchmark_id=context.market_benchmark.instrument_id if context.market_benchmark else None)
138 result.benchmark_mapping_version = context.mapping_version
139 if not classification_valid:
140 result.missing_inputs.append("AUTHORITATIVE_SECTOR_CLASSIFICATION")
141 stock_dates, stock_conflict, stock_stale, source = _dated_history(instrument_id, stock_history,
142 daily.get(instrument_id, ()), as_of, currency, trusted_providers, cfg.max_age_days, stock=True)
143 result.history_sources['stock'] = source
144 if stock_stale:
145 result.stale_inputs.append("STOCK_HISTORY")
146 if stock_conflict:
147 result.missing_inputs.append("CONFLICTING_STOCK_PRICE")
148 benchmark_data = {}
149 for name, reference in (("sector", context.sector_benchmark if classification_valid else None), ("market", context.market_benchmark)):
150 if reference is None:
151 result.missing_inputs.append(f"{name.upper()}_BENCHMARK_MAPPING")
152 result.benchmark_states[name] = ('NO_SECTOR_CLASSIFICATION' if name == 'sector' and not classification_valid
153 else getattr(context, f'{name}_mapping_status') or
154 ('UNMAPPED_SECTOR_BENCHMARK' if name == 'sector' else 'BENCHMARK_IDENTITY_UNAVAILABLE'))
155 continue
156 if reference.instrument_id == instrument_id or not reference.currency:
157 result.missing_inputs.append(f"INVALID_{name.upper()}_BENCHMARK_MAPPING")
158 result.benchmark_states[name] = 'BENCHMARK_IDENTITY_UNAVAILABLE'
159 continue
160 dates, conflict, stale, source = _dated_history(reference.instrument_id, histories.get(reference.instrument_id, ()),
161 daily.get(reference.instrument_id, ()), as_of, reference.currency, reference.trusted_providers, cfg.max_age_days)
162 result.history_sources[name] = source
163 if not dates or conflict:
164 result.missing_inputs.append(f"{name.upper()}_HISTORY" if not conflict else f"CONFLICTING_{name.upper()}_PRICE")
165 result.benchmark_states[name] = 'BENCHMARK_HISTORY_UNAVAILABLE'
166 continue
167 if stale:
168 result.stale_inputs.append(f"{name.upper()}_HISTORY")
169 result.benchmark_states[name] = 'STALE_BENCHMARK_HISTORY'
170 continue
171 benchmark_data[name] = dates
172 result.benchmark_states[name] = 'INSUFFICIENT_OVERLAP'
173 edges = {}
174 for suffix, lookback, weight in zip(("1_w", "1_m", "3_m", "6_m"), cfg.return_lookbacks, cfg.period_weights):
175 if not stock_conflict and len(stock_dates) > lookback:
176 ordered = sorted(stock_dates)
177 dates = ordered[-lookback-1], ordered[-1]
178 stock_return = percentage(stock_dates[dates[1]], stock_dates[dates[0]])
179 setattr(result, f"stock_return{suffix}", stock_return)
180 result.comparison_windows[suffix.replace("_", "").upper()] = dates
181 period_edges = []
182 for name in ("sector", "market"):
183 rows = benchmark_data.get(name, {})
184 if not stock_stale and dates[0] in rows and dates[1] in rows:
185 value = percentage(rows[dates[1]], rows[dates[0]])
186 setattr(result, f"{name}_return{suffix}", value)
187 setattr(result, f"relative_vs_{name}{suffix}", stock_return - value)
188 period_edges.append((stock_return - value) / lookback)
189 result.benchmark_states[name] = 'AVAILABLE'
190 elif rows and not stock_stale:
191 result.missing_inputs.append(f"{name.upper()}_ALIGNED_DATES_{suffix.replace('_', '').upper()}")
192 if period_edges:
193 edges[suffix] = (sum(period_edges) / len(period_edges), weight)
194 # Require at least two horizons to claim consistency. Missing benchmark
195 # legs are omitted, never assigned a zero relative return.
196 if len(edges) >= 2 and sum(weight for _, weight in edges.values()) > 0:
197 average = sum(edge * weight for edge, weight in edges.values()) / sum(weight for _, weight in edges.values())
198 result.relative_strength_score = min(100, max(0, 50 + 50 * average / cfg.full_scale_edge_per_observation_pct))
199 # sectorState describes the stock's relative leadership with sector
200 # context. Market-only evidence may score partially, but is not a sector state.
201 sector_periods = sum(getattr(result, f"relative_vs_sector{s}") is not None for s in ("1_w", "1_m", "3_m", "6_m"))
202 if sector_periods >= 2:
203 values = [v for v, _ in edges.values()]
204 if all(v > cfg.material_edge_per_observation_pct for v in values):
205 result.sector_state = "LEADING"
206 elif all(v < -cfg.material_edge_per_observation_pct for v in values):
207 result.sector_state = "LAGGING"
208 else:
209 short = [edges[s][0] for s in ("1_w", "1_m") if s in edges]
210 long = [edges[s][0] for s in ("3_m", "6_m") if s in edges]
211 change = sum(short)/len(short) - sum(long)/len(long) if short and long else 0
212 result.sector_state = ("IMPROVING" if change > cfg.improvement_per_observation_pct else
213 "WEAKENING" if change < -cfg.improvement_per_observation_pct else "NEUTRAL")
214 relative_fields = []
215 for name, field in SectorRelativeStrengthSnapshot.model_fields.items():
216 if name.startswith(("stock_return", "sector_return", "market_return", "relative_vs_")):
217 value = getattr(result, name)
218 alias = field.alias or name
219 result.feature_states[alias] = "MISSING" if value is None else "STALE" if stock_stale else "AVAILABLE"
220 if value is None:
221 result.missing_inputs.append(alias)
222 else:
223 setattr(result, name, round(value, 8))
224 if name.startswith("relative_vs_"):
225 relative_fields.append(value)
226 result.confidence = round(100 * sum(v is not None for v in relative_fields) / 8, 6)
227 if result.relative_strength_score is not None:
228 result.relative_strength_score = round(result.relative_strength_score, 8)
229 result.missing_inputs = sorted(set(result.missing_inputs))
230 result.stale_inputs = sorted(set(result.stale_inputs))
231 return result
232
233
234 def _dated_history(key, closes, bars, as_of, currency, providers, max_age, stock=False):
235 """Keep daily DATEs intact; no timestamp shift, filling, or nearest-date match."""
236 history, present = normalize_daily_history(key, bars, as_of=as_of, currency=currency, trusted_providers=providers)
237 today = utc(as_of).astimezone(ZoneInfo('Asia/Kolkata')).date()
238 stale = bool(history.observations) and today - history.observations[-1].trading_date > timedelta(days=max_age)
239 fallback = None
240 if present and stock and not history.current_conflict and (len(history.observations) < 20 or stale):
241 fallback = normalize_price_history(key, closes, as_of=as_of, currency=currency, trusted_providers=providers)
242 if (len(fallback.observations) >= 20 and not fallback.current_conflict and
243 utc(as_of) - utc(fallback.observations[-1].observed_at) <= timedelta(days=max_age)):
244 present = False
245 if present:
246 return ({row.trading_date:float(row.close) for row in history.observations}, history.current_conflict,
247 stale, 'DAILY_MARKET_BAR_NSE')
248 fallback = fallback or normalize_price_history(key, closes, as_of=as_of, currency=currency, trusted_providers=providers)
249 stale = bool(fallback.observations) and utc(as_of) - utc(fallback.observations[-1].observed_at) > timedelta(days=max_age)
250 return ({utc(row.observed_at).date():float(row.price) for row in fallback.observations},
251 fallback.current_conflict, stale, 'CLOSE_ONLY_FALLBACK')