|
|
"""Synthetic Yahoo Top Gainer list reconstruction from intraday bar data.
|
|
|
|
|
|
Since Yahoo Finance does not provide historical day_gainers records, we
|
|
|
reconstruct a *synthetic* ranking at each bar tick during the collection window
|
|
|
by computing each ticker's percent change from previous close using its
|
|
|
intraday bars. This mirrors the Yahoo day_gainers screener filter:
|
|
|
- percent_change > 3% (we use the caller-supplied min filter)
|
|
|
- US equities (guaranteed by universe)
|
|
|
- market cap / dollar volume surrogate filter applied by caller
|
|
|
"""
|
|
|
from __future__ import annotations
|
|
|
|
|
|
import datetime as dt
|
|
|
from typing import Any
|
|
|
from zoneinfo import ZoneInfo
|
|
|
|
|
|
_ET_ZONE = ZoneInfo("America/New_York")
|
|
|
|
|
|
|
|
|
def _parse_ts(ts_str: str) -> dt.datetime:
|
|
|
"""Parse ISO timestamp to naive UTC datetime for comparison."""
|
|
|
import dateutil.parser as dparse
|
|
|
parsed = dparse.parse(ts_str)
|
|
|
if parsed.tzinfo is not None:
|
|
|
parsed = parsed.astimezone(dt.timezone.utc).replace(tzinfo=None)
|
|
|
return parsed
|
|
|
|
|
|
|
|
|
def _bar_ts_naive_utc(bar: dict) -> dt.datetime:
|
|
|
return _parse_ts(bar["timestamp"])
|
|
|
|
|
|
|
|
|
def _et_to_utc_naive(date: dt.date, hour: int, minute: int) -> dt.datetime:
|
|
|
"""Convert an ET time on a given date to a naive UTC datetime (DST-aware)."""
|
|
|
et_aware = dt.datetime(date.year, date.month, date.day, hour, minute,
|
|
|
tzinfo=_ET_ZONE)
|
|
|
return et_aware.astimezone(dt.timezone.utc).replace(tzinfo=None)
|
|
|
|
|
|
|
|
|
# ── Snapshot reconstruction ────────────────────────────────────────────────────
|
|
|
|
|
|
# Default V1 collection ticks (09:30–09:55 ET, 5-min bars)
|
|
|
_COLLECTION_TICKS_ET_V1 = [
|
|
|
dt.time(9, 30), dt.time(9, 35), dt.time(9, 40),
|
|
|
dt.time(9, 45), dt.time(9, 50), dt.time(9, 55),
|
|
|
]
|
|
|
|
|
|
# Default V2 collection ticks (09:35–10:15 ET, 1-min bars)
|
|
|
_COLLECTION_TICKS_ET_V2 = [
|
|
|
dt.time(9, 35), dt.time(9, 45), dt.time(10, 0), dt.time(10, 15),
|
|
|
]
|
|
|
|
|
|
|
|
|
def reconstruct_gainer_snapshots(
|
|
|
bars_by_symbol: dict[str, list[dict]],
|
|
|
prev_closes: dict[str, float],
|
|
|
date_str: str,
|
|
|
min_pct_change: float = 0.03,
|
|
|
top_n: int = 100,
|
|
|
collection_ticks_et: list[dt.time] | None = None,
|
|
|
) -> list[dict[str, Any]]:
|
|
|
"""Build synthetic snapshot rows for each tick during the collection window.
|
|
|
|
|
|
Args:
|
|
|
bars_by_symbol: {symbol: [bar_dict, ...]} from IntradayCache.
|
|
|
prev_closes: {symbol: prev_close_price} from DailyBarCache.
|
|
|
date_str: 'YYYY-MM-DD'
|
|
|
min_pct_change: minimum gain to appear (Yahoo ~3%).
|
|
|
top_n: maximum symbols per tick.
|
|
|
collection_ticks_et: list of ET times to reconstruct rankings at.
|
|
|
Defaults to V1 ticks (09:30–09:55 every 5m).
|
|
|
Pass _COLLECTION_TICKS_ET_V2 for V2 dataset builds.
|
|
|
|
|
|
Returns:
|
|
|
List of snapshot dicts: {captured_at, symbol, rank, price, pct_change, volume, market_cap}
|
|
|
captured_at is an ISO string corresponding to the tick timestamp.
|
|
|
market_cap is None (not available in bar data; use enrichment surrogate).
|
|
|
"""
|
|
|
ticks = collection_ticks_et if collection_ticks_et is not None else _COLLECTION_TICKS_ET_V1
|
|
|
date = dt.date.fromisoformat(date_str)
|
|
|
rows: list[dict[str, Any]] = []
|
|
|
|
|
|
for tick_time in ticks:
|
|
|
tick_cutoff_utc = _et_to_utc_naive(date, tick_time.hour, tick_time.minute)
|
|
|
tick_dt_et = dt.datetime(date.year, date.month, date.day,
|
|
|
tick_time.hour, tick_time.minute, 0)
|
|
|
tick_iso = tick_dt_et.strftime("%Y-%m-%dT%H:%M:00")
|
|
|
|
|
|
candidates: list[dict[str, Any]] = []
|
|
|
for sym, bars in bars_by_symbol.items():
|
|
|
prev_close = prev_closes.get(sym)
|
|
|
if not prev_close or prev_close <= 0:
|
|
|
continue
|
|
|
|
|
|
# Find the latest bar whose timestamp <= tick cutoff
|
|
|
latest_bar: dict | None = None
|
|
|
for b in bars:
|
|
|
b_ts = _bar_ts_naive_utc(b)
|
|
|
if b_ts <= tick_cutoff_utc:
|
|
|
latest_bar = b
|
|
|
else:
|
|
|
break # bars are sorted ascending
|
|
|
|
|
|
if latest_bar is None:
|
|
|
continue
|
|
|
|
|
|
price = float(latest_bar["close"])
|
|
|
if price <= 0:
|
|
|
continue
|
|
|
pct_change = (price - prev_close) / prev_close
|
|
|
if pct_change < min_pct_change:
|
|
|
continue
|
|
|
|
|
|
# Volume so far
|
|
|
cum_vol = sum(
|
|
|
float(b.get("volume", 0) or 0)
|
|
|
for b in bars
|
|
|
if _bar_ts_naive_utc(b) <= tick_cutoff_utc
|
|
|
)
|
|
|
|
|
|
candidates.append({
|
|
|
"symbol": sym,
|
|
|
"price": price,
|
|
|
"pct_change": pct_change,
|
|
|
"volume": cum_vol,
|
|
|
"market_cap": None, # not available; caller applies dollar-vol filter
|
|
|
})
|
|
|
|
|
|
# Sort descending by pct_change, take top_n
|
|
|
candidates.sort(key=lambda c: c["pct_change"], reverse=True)
|
|
|
for rank, cand in enumerate(candidates[:top_n], start=1):
|
|
|
rows.append({
|
|
|
"captured_at": tick_iso,
|
|
|
"symbol": cand["symbol"],
|
|
|
"rank": rank,
|
|
|
"price": cand["price"],
|
|
|
"pct_change": cand["pct_change"],
|
|
|
"volume": cand["volume"],
|
|
|
"market_cap": cand["market_cap"],
|
|
|
})
|
|
|
|
|
|
return rows
|