Keysat
97 lines
4.2 KiB
Python
97 lines
4.2 KiB
Python
"""External-confirmation data for the resolver (DESIGN_v2 §1). Price series via FMP (already paid for).
|
|
|
|
This is the *resolving* leg (§6.2): real-world repricing, not discourse. Kept deliberately simple and
|
|
transparent — the resolution rule is pre-registered, so the code here only fetches + applies it.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import requests
|
|
|
|
_FMP = "https://financialmodelingprep.com"
|
|
|
|
|
|
def fetch_eod(api_key: str, symbol: str, start: str, end: str) -> list[tuple[str, float]]:
|
|
"""Daily (date, close) for a symbol. Tries the FMP 'stable' then legacy 'v3' price endpoints."""
|
|
s = requests.Session()
|
|
attempts = [
|
|
(f"{_FMP}/stable/historical-price-eod/full", {"symbol": symbol, "from": start, "to": end}),
|
|
(f"{_FMP}/api/v3/historical-price-full/{symbol}", {"from": start, "to": end}),
|
|
]
|
|
for url, params in attempts:
|
|
try:
|
|
r = s.get(url, params={**params, "apikey": api_key}, timeout=40)
|
|
if r.status_code != 200:
|
|
continue
|
|
j = r.json()
|
|
except Exception: # noqa: BLE001
|
|
continue
|
|
rows = j.get("historical") if isinstance(j, dict) else j
|
|
if not rows:
|
|
continue
|
|
out = [(x["date"][:10], x.get("close") or x.get("adjClose")) for x in rows
|
|
if x.get("date") and (x.get("close") or x.get("adjClose"))]
|
|
if out:
|
|
return sorted(out)
|
|
return []
|
|
|
|
|
|
def basket_index(prices_by_symbol: dict[str, list[tuple[str, float]]]) -> list[tuple[str, float]]:
|
|
"""Equal-weight, each-symbol-normalized-to-its-own-first-close index, averaged over dates where
|
|
data exists. (Symbols that IPO'd mid-window enter at 1.0 when they start — flagged by the caller.)"""
|
|
norm = {}
|
|
for sym, series in prices_by_symbol.items():
|
|
if series:
|
|
base = series[0][1]
|
|
norm[sym] = {d: c / base for d, c in series if base}
|
|
dates = sorted({d for n in norm.values() for d in n})
|
|
idx = []
|
|
for d in dates:
|
|
vals = [n[d] for n in norm.values() if d in n]
|
|
if vals:
|
|
idx.append((d, sum(vals) / len(vals)))
|
|
return idx
|
|
|
|
|
|
def index_value_at(index: list[tuple[str, float]], date: str | None) -> float | None:
|
|
"""Latest index value on or before `date` (baseline if the signal predates the data)."""
|
|
if not index or not date:
|
|
return None
|
|
vals = [v for d, v in index if d <= date]
|
|
return vals[-1] if vals else index[0][1]
|
|
|
|
|
|
def runway_at_signal(index: list[tuple[str, float]], signal_date: str | None) -> float | None:
|
|
"""Fraction of the durable move STILL AHEAD at the signal date (DESIGN_v2.1 Correction A).
|
|
1.0 = whole move ahead (signal before it); 0.0 = signal at the peak. The right metric for a
|
|
long-duration holder — a modestly-late signal with most of the move ahead is still actionable."""
|
|
if not index or not signal_date:
|
|
return None
|
|
base = index[0][1]
|
|
peak = max(v for _, v in index)
|
|
val = index_value_at(index, signal_date)
|
|
if peak <= base or val is None:
|
|
return None
|
|
return round(max(0.0, (peak - val) / (peak - base)), 2)
|
|
|
|
|
|
def resolve_reprice(index: list[tuple[str, float]], *, threshold_pct: float, hold_pct: float,
|
|
hold_days: int) -> dict:
|
|
"""Apply the pre-registered rule: first date the index is ≥ +threshold% vs baseline AND still
|
|
≥ +hold% `hold_days` later. Returns {confirmed, repricing_date, peak_pct}."""
|
|
from datetime import datetime, timedelta
|
|
if not index:
|
|
return {"confirmed": False, "repricing_date": None, "peak_pct": None}
|
|
base = index[0][1]
|
|
thr = 1.0 + threshold_pct / 100.0
|
|
hold = 1.0 + hold_pct / 100.0
|
|
by_date = dict(index)
|
|
dates = [d for d, _ in index]
|
|
peak = max(v for _, v in index)
|
|
for d, v in index:
|
|
if v / base >= thr:
|
|
target = (datetime.strptime(d, "%Y-%m-%d") + timedelta(days=hold_days)).strftime("%Y-%m-%d")
|
|
later = [vv for dd, vv in index if dd >= target]
|
|
if later and (later[0] / base) >= hold:
|
|
return {"confirmed": True, "repricing_date": d, "peak_pct": round((peak / base - 1) * 100, 1)}
|
|
return {"confirmed": False, "repricing_date": None, "peak_pct": round((peak / base - 1) * 100, 1)}
|