| 1 | """Pure market-performance projection for the global sector dashboard.""" |
| 2 | from __future__ import annotations |
| 3 | |
| 4 | from datetime import timedelta |
| 5 | from decimal import Decimal |
| 6 | from typing import Iterable |
| 7 | |
| 8 | |
| 9 | _PERIOD_DAYS = {"WEEK": 7, "MONTH": 30, "YEAR": 365} |
| 10 | _EUROPE_COUNTRIES = {"AT", "BE", "CH", "DE", "DK", "ES", "FI", "FR", "GB", "IE", "IT", "NL", "NO", "PT", "SE"} |
| 11 | |
| 12 | |
| 13 | def belongs_to_region(item: dict, region: str) -> bool: |
| 14 | region = region.upper() |
| 15 | country = str(item.get("country") or "").upper() |
| 16 | exchange = str(item.get("exchange") or item.get("mic") or "").upper() |
| 17 | if region == "INDIA": |
| 18 | return country in {"IN", "IND", "INDIA"} or exchange in {"XNSE", "NSE", "XBOM", "BSE"} |
| 19 | if region == "USA": |
| 20 | return country in {"US", "USA"} or exchange in {"XNAS", "XNYS", "ARCX", "BATS"} |
| 21 | if region == "EUROPE": |
| 22 | return country in _EUROPE_COUNTRIES or exchange in {"XETR", "XAMS", "AEB", "XLON", "XPAR", "XSWX", "XMIL", "XMAD", "XSTO", "XHEL", "XCSE", "XOSL"} |
| 23 | return False |
| 24 | |
| 25 | |
| 26 | def performance_window(observations: Iterable, period: str): |
| 27 | """Return latest and the nearest valid prior observation for a period.""" |
| 28 | period = period.upper() |
| 29 | if period not in {"DAY", *_PERIOD_DAYS}: |
| 30 | raise ValueError("UNSUPPORTED_PERFORMANCE_PERIOD") |
| 31 | usable = sorted((value for value in observations if value.price is not None and value.price > 0), key=lambda value: value.observed_at) |
| 32 | if len(usable) < 2: |
| 33 | return None |
| 34 | latest = usable[-1] |
| 35 | # A day means the preceding *observed trading close*, rather than a |
| 36 | # calendar-day subtraction. This makes a Monday correctly compare with |
| 37 | # Friday (or the preceding holiday-adjusted observation). |
| 38 | if period == "DAY": |
| 39 | reference = usable[-2] |
| 40 | else: |
| 41 | target = latest.observed_at - timedelta(days=_PERIOD_DAYS[period]) |
| 42 | reference = next((value for value in reversed(usable[:-1]) if value.observed_at <= target), None) |
| 43 | if reference is None or reference.price <= 0: |
| 44 | return None |
| 45 | return latest, reference, (latest.price - reference.price) / reference.price * Decimal("100") |
| 46 | |
| 47 | |
| 48 | def deduplicated_performance_candidates(candidates: Iterable[dict]) -> list[dict]: |
| 49 | """Resolve duplicate global identities before either performance ranking.""" |
| 50 | winners: dict[str, dict] = {} |
| 51 | for candidate in candidates: |
| 52 | identity = str(candidate["globalInstrumentId"]) |
| 53 | previous = winners.get(identity) |
| 54 | key = (-candidate["performancePct"], str(candidate.get("ticker") or "").upper(), identity) |
| 55 | if previous is None or key < (-previous["performancePct"], str(previous.get("ticker") or "").upper(), identity): |
| 56 | winners[identity] = candidate |
| 57 | return list(winners.values()) |
| 58 | |
| 59 | |
| 60 | def rank_performers(candidates: Iterable[dict], limit: int = 5) -> tuple[list[dict], list[dict]]: |
| 61 | """Return deterministic top and worst performers from one read-only input.""" |
| 62 | rows = deduplicated_performance_candidates(candidates) |
| 63 | bounded = max(1, min(5, limit)) |
| 64 | ticker_key = lambda value: (str(value.get("ticker") or "").upper(), str(value["globalInstrumentId"])) |
| 65 | best = sorted(rows, key=lambda value: (-value["performancePct"], *ticker_key(value)))[:bounded] |
| 66 | worst = sorted(rows, key=lambda value: (value["performancePct"], *ticker_key(value)))[:bounded] |
| 67 | return best, worst |
| 68 | |
| 69 | |
| 70 | def rank_top_gainers(candidates: Iterable[dict], limit: int = 5) -> list[dict]: |
| 71 | """Compatibility helper retained for callers of the earlier endpoint.""" |
| 72 | return rank_performers(candidates, limit)[0] |