You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
251 lines
8.9 KiB
Python
251 lines
8.9 KiB
Python
"""Utilities for reconstructing daily top-gainer style signals from bars.
|
|
|
|
This module is intentionally generic. A Yahoo-style probe can reuse it by
|
|
ranking candidates on signal-day return after applying liquidity/quality gates.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import statistics
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DailyGainerCriteria:
|
|
min_change: float = 0.08
|
|
max_change: float | None = 0.25
|
|
min_volume_ratio: float = 1.5
|
|
min_close_location: float = 0.60
|
|
min_avg_dollar_volume: float = 10_000_000.0
|
|
max_gap: float | None = None
|
|
top_n: int = 3
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DailyGainerSignal:
|
|
symbol: str
|
|
signal_date: dt.date
|
|
rank: int
|
|
signal_return: float
|
|
volume_ratio_20d: float
|
|
close_location: float
|
|
gap_size: float
|
|
avg_dollar_volume_20d: float
|
|
signal_close: float
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DailyGainerTrade:
|
|
symbol: str
|
|
signal_date: dt.date
|
|
rank: int
|
|
signal_return: float
|
|
volume_ratio_20d: float
|
|
close_location: float
|
|
gap_size: float
|
|
avg_dollar_volume_20d: float
|
|
signal_close: float
|
|
entry_date: dt.date
|
|
entry_price: float
|
|
entry_gap_pct: float
|
|
hold_1d_return: float | None = None
|
|
hold_3d_return: float | None = None
|
|
hold_5d_return: float | None = None
|
|
|
|
|
|
def collect_daily_gainer_signals(
|
|
*,
|
|
bars_by_symbol: dict[str, dict[dt.date, dict[str, Any]]],
|
|
start_date: dt.date,
|
|
end_date: dt.date,
|
|
criteria: DailyGainerCriteria,
|
|
) -> dict[dt.date, list[DailyGainerSignal]]:
|
|
"""Return top-N daily gainer signals per date from historical bars."""
|
|
candidates_by_date: dict[dt.date, list[DailyGainerSignal]] = {}
|
|
|
|
for raw_symbol, bars in bars_by_symbol.items():
|
|
symbol = str(raw_symbol or "").upper()
|
|
if not symbol or len(bars) < 21:
|
|
continue
|
|
|
|
sorted_dates = sorted(bars)
|
|
for idx in range(20, len(sorted_dates)):
|
|
signal_date = sorted_dates[idx]
|
|
if signal_date < start_date or signal_date > end_date:
|
|
continue
|
|
|
|
prev_date = sorted_dates[idx - 1]
|
|
prev_bar = bars[prev_date]
|
|
signal_bar = bars[signal_date]
|
|
|
|
prev_close = float(prev_bar.get("close") or 0.0)
|
|
signal_open = float(signal_bar.get("open") or 0.0)
|
|
signal_high = float(signal_bar.get("high") or 0.0)
|
|
signal_low = float(signal_bar.get("low") or 0.0)
|
|
signal_close = float(signal_bar.get("close") or 0.0)
|
|
signal_volume = float(signal_bar.get("volume") or 0.0)
|
|
|
|
if prev_close <= 0 or signal_open <= 0 or signal_close <= 0:
|
|
continue
|
|
|
|
lookback_dates = sorted_dates[idx - 20 : idx]
|
|
avg_volume = statistics.fmean(
|
|
float(bars[d].get("volume") or 0.0)
|
|
for d in lookback_dates
|
|
)
|
|
avg_dollar_volume = statistics.fmean(
|
|
float(bars[d].get("close") or 0.0) * float(bars[d].get("volume") or 0.0)
|
|
for d in lookback_dates
|
|
)
|
|
volume_ratio = signal_volume / avg_volume if avg_volume > 0 else 0.0
|
|
signal_return = signal_close / prev_close - 1.0
|
|
gap_size = signal_open / prev_close - 1.0
|
|
intraday_range = signal_high - signal_low
|
|
close_location = (
|
|
(signal_close - signal_low) / intraday_range
|
|
if intraday_range > 0
|
|
else 0.5
|
|
)
|
|
|
|
if signal_return < criteria.min_change:
|
|
continue
|
|
if criteria.max_change is not None and signal_return > criteria.max_change:
|
|
continue
|
|
if volume_ratio < criteria.min_volume_ratio:
|
|
continue
|
|
if close_location < criteria.min_close_location:
|
|
continue
|
|
if avg_dollar_volume < criteria.min_avg_dollar_volume:
|
|
continue
|
|
if criteria.max_gap is not None and gap_size > criteria.max_gap:
|
|
continue
|
|
|
|
candidates_by_date.setdefault(signal_date, []).append(
|
|
DailyGainerSignal(
|
|
symbol=symbol,
|
|
signal_date=signal_date,
|
|
rank=0,
|
|
signal_return=signal_return,
|
|
volume_ratio_20d=volume_ratio,
|
|
close_location=close_location,
|
|
gap_size=gap_size,
|
|
avg_dollar_volume_20d=avg_dollar_volume,
|
|
signal_close=signal_close,
|
|
)
|
|
)
|
|
|
|
ranked_by_date: dict[dt.date, list[DailyGainerSignal]] = {}
|
|
for signal_date, signals in candidates_by_date.items():
|
|
ranked = sorted(
|
|
signals,
|
|
key=lambda item: (
|
|
-item.signal_return,
|
|
-item.volume_ratio_20d,
|
|
-item.avg_dollar_volume_20d,
|
|
item.symbol,
|
|
),
|
|
)
|
|
trimmed: list[DailyGainerSignal] = []
|
|
for rank, signal in enumerate(ranked[: criteria.top_n], start=1):
|
|
trimmed.append(
|
|
DailyGainerSignal(
|
|
symbol=signal.symbol,
|
|
signal_date=signal.signal_date,
|
|
rank=rank,
|
|
signal_return=signal.signal_return,
|
|
volume_ratio_20d=signal.volume_ratio_20d,
|
|
close_location=signal.close_location,
|
|
gap_size=signal.gap_size,
|
|
avg_dollar_volume_20d=signal.avg_dollar_volume_20d,
|
|
signal_close=signal.signal_close,
|
|
)
|
|
)
|
|
if trimmed:
|
|
ranked_by_date[signal_date] = trimmed
|
|
|
|
return ranked_by_date
|
|
|
|
|
|
def evaluate_daily_gainer_signals(
|
|
*,
|
|
signals_by_date: dict[dt.date, list[DailyGainerSignal]],
|
|
bars_by_symbol: dict[str, dict[dt.date, dict[str, Any]]],
|
|
horizons: tuple[int, ...] = (1, 3, 5),
|
|
) -> list[DailyGainerTrade]:
|
|
"""Evaluate next-open continuation returns for reconstructed gainer signals."""
|
|
sorted_dates_cache: dict[str, list[dt.date]] = {}
|
|
date_index_cache: dict[str, dict[dt.date, int]] = {}
|
|
trades: list[DailyGainerTrade] = []
|
|
|
|
for signal_date in sorted(signals_by_date):
|
|
for signal in signals_by_date[signal_date]:
|
|
symbol = signal.symbol.upper()
|
|
bars = bars_by_symbol.get(symbol) or {}
|
|
if len(bars) < 2:
|
|
continue
|
|
|
|
sorted_dates = sorted_dates_cache.get(symbol)
|
|
if sorted_dates is None:
|
|
sorted_dates = sorted(bars)
|
|
sorted_dates_cache[symbol] = sorted_dates
|
|
date_index_cache[symbol] = {d: idx for idx, d in enumerate(sorted_dates)}
|
|
|
|
date_index = date_index_cache[symbol]
|
|
signal_idx = date_index.get(signal.signal_date)
|
|
if signal_idx is None or signal_idx + 1 >= len(sorted_dates):
|
|
continue
|
|
|
|
entry_date = sorted_dates[signal_idx + 1]
|
|
entry_bar = bars.get(entry_date) or {}
|
|
entry_price = float(entry_bar.get("open") or 0.0)
|
|
if entry_price <= 0:
|
|
continue
|
|
|
|
returns: dict[int, float | None] = {h: None for h in horizons}
|
|
for horizon in horizons:
|
|
exit_idx = signal_idx + horizon
|
|
if exit_idx >= len(sorted_dates):
|
|
continue
|
|
exit_date = sorted_dates[exit_idx]
|
|
exit_bar = bars.get(exit_date) or {}
|
|
exit_close = float(exit_bar.get("close") or 0.0)
|
|
if exit_close <= 0:
|
|
continue
|
|
returns[horizon] = exit_close / entry_price - 1.0
|
|
|
|
trades.append(
|
|
DailyGainerTrade(
|
|
symbol=symbol,
|
|
signal_date=signal.signal_date,
|
|
rank=signal.rank,
|
|
signal_return=signal.signal_return,
|
|
volume_ratio_20d=signal.volume_ratio_20d,
|
|
close_location=signal.close_location,
|
|
gap_size=signal.gap_size,
|
|
avg_dollar_volume_20d=signal.avg_dollar_volume_20d,
|
|
signal_close=signal.signal_close,
|
|
entry_date=entry_date,
|
|
entry_price=entry_price,
|
|
entry_gap_pct=entry_price / signal.signal_close - 1.0,
|
|
hold_1d_return=returns.get(1),
|
|
hold_3d_return=returns.get(3),
|
|
hold_5d_return=returns.get(5),
|
|
)
|
|
)
|
|
|
|
return trades
|
|
|
|
|
|
def summarize_metric(values: list[float]) -> dict[str, float | int | None]:
|
|
"""Return compact summary stats for a metric series."""
|
|
if not values:
|
|
return {"count": 0, "mean": None, "median": None, "win_rate": None}
|
|
return {
|
|
"count": len(values),
|
|
"mean": statistics.fmean(values),
|
|
"median": statistics.median(values),
|
|
"win_rate": sum(1 for value in values if value > 0) / len(values),
|
|
}
|
|
|