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.
456 lines
16 KiB
Python
456 lines
16 KiB
Python
"""Probe Yahoo-style daily gainer continuation using reconstructed price signals.
|
|
|
|
This is not an exact Yahoo backtest. It rebuilds the same class of setup from
|
|
historical daily bars:
|
|
- strong signal-day return
|
|
- strong volume ratio
|
|
- strong close location
|
|
- liquid universe
|
|
- ranked by signal-day return
|
|
|
|
The goal is hypothesis validation, not final strategy certification.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import csv
|
|
import datetime as dt
|
|
import json
|
|
import sys
|
|
from dataclasses import asdict
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
from libs.backtest.daily_gainers import (
|
|
DailyGainerCriteria,
|
|
DailyGainerTrade,
|
|
collect_daily_gainer_signals,
|
|
evaluate_daily_gainer_signals,
|
|
summarize_metric,
|
|
)
|
|
from libs.backtest.snapshot_store import SnapshotStore
|
|
from libs.common.config import get_settings
|
|
from libs.common.logging import configure_logging
|
|
from libs.common.time_utils import is_trading_day, previous_trading_day, to_eastern, utc_now
|
|
from libs.oracle_client import ScreenerService, make_oracle_client
|
|
|
|
|
|
UNIVERSE_PROFILES: dict[str, dict[str, Any]] = {
|
|
"midlarge-liquid-long-v1": {
|
|
"market_cap_min": 2_000_000_000,
|
|
"price_min": 15,
|
|
"exchange": "NYSE,NASDAQ,AMEX",
|
|
"exclude_types": "ETF,FUND,ADR,SPAC",
|
|
},
|
|
"midplus-liquid-long-v1": {
|
|
"market_cap_min": 1_500_000_000,
|
|
"price_min": 12,
|
|
"exchange": "NYSE,NASDAQ,AMEX",
|
|
"exclude_types": "ETF,FUND,ADR,SPAC",
|
|
},
|
|
"midwide-liquid-long-v1": {
|
|
"market_cap_min": 1_000_000_000,
|
|
"price_min": 10,
|
|
"exchange": "NYSE,NASDAQ,AMEX",
|
|
"exclude_types": "ETF,FUND,ADR,SPAC",
|
|
},
|
|
"smallcap-liquid-long-v1": {
|
|
"market_cap_min": 500_000_000,
|
|
"market_cap_max": 2_000_000_000,
|
|
"price_min": 8,
|
|
"exchange": "NYSE,NASDAQ,AMEX",
|
|
"exclude_types": "ETF,FUND,ADR,SPAC",
|
|
},
|
|
}
|
|
|
|
PROFILE_SYMBOL_FALLBACK_FILES: dict[str, str] = {
|
|
"midlarge-liquid-long-v1": "configs/symbols_midlarge_snapshot_exact.yaml",
|
|
"midplus-liquid-long-v1": "configs/symbols_midlarge_snapshot_exact.yaml",
|
|
"midwide-liquid-long-v1": "configs/symbols_midcap.yaml",
|
|
"smallcap-liquid-long-v1": "configs/symbols_smallmid.yaml",
|
|
}
|
|
|
|
|
|
def _default_end_date() -> dt.date:
|
|
now_et = to_eastern(utc_now())
|
|
if not is_trading_day(now_et.date()) or now_et.hour >= 16:
|
|
return now_et.date()
|
|
return previous_trading_day(now_et.date())
|
|
|
|
|
|
def _default_start_date(end_date: dt.date) -> dt.date:
|
|
return end_date - dt.timedelta(days=730)
|
|
|
|
|
|
def _parse_date(value: str, *, is_end: bool = False) -> dt.date:
|
|
parts = value.split("-")
|
|
if len(parts) == 1 and value.isdigit() and len(value) == 4:
|
|
year = int(value)
|
|
return dt.date(year, 12, 31) if is_end else dt.date(year, 1, 1)
|
|
if len(parts) == 2 and all(part.isdigit() for part in parts):
|
|
year = int(parts[0])
|
|
month = int(parts[1])
|
|
if is_end:
|
|
if month == 12:
|
|
return dt.date(year, 12, 31)
|
|
return dt.date(year, month + 1, 1) - dt.timedelta(days=1)
|
|
return dt.date(year, month, 1)
|
|
return dt.date.fromisoformat(value)
|
|
|
|
|
|
def _load_symbols_from_yaml(path: Path) -> list[str]:
|
|
payload = yaml.safe_load(path.read_text()) or {}
|
|
symbols = payload.get("symbols", [])
|
|
return sorted({str(symbol).upper() for symbol in symbols if symbol})
|
|
|
|
|
|
async def _resolve_profile_symbols(universe_profile: str) -> list[str]:
|
|
profile = UNIVERSE_PROFILES.get(universe_profile)
|
|
if profile is None:
|
|
raise ValueError(
|
|
f"Unknown universe profile '{universe_profile}'. "
|
|
f"Available: {sorted(UNIVERSE_PROFILES)}"
|
|
)
|
|
async with make_oracle_client() as client:
|
|
screener = ScreenerService(client)
|
|
stocks = await screener.search_all_stocks(
|
|
market_cap_min=profile.get("market_cap_min"),
|
|
market_cap_max=profile.get("market_cap_max"),
|
|
exchange=profile["exchange"],
|
|
exclude_types=profile["exclude_types"],
|
|
price_min=profile["price_min"],
|
|
)
|
|
return sorted({str(stock.symbol).upper() for stock in stocks if stock.symbol})
|
|
|
|
|
|
def _resolve_symbols(args: argparse.Namespace) -> tuple[list[str], str]:
|
|
if args.symbol:
|
|
symbols = sorted({str(symbol).upper() for symbol in args.symbol if symbol})
|
|
return symbols, "explicit"
|
|
|
|
if args.symbols_file:
|
|
path = Path(args.symbols_file)
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"symbols file not found: {path}")
|
|
return _load_symbols_from_yaml(path), f"yaml:{path}"
|
|
|
|
profile = args.universe_profile or "midlarge-liquid-long-v1"
|
|
try:
|
|
symbols = asyncio.run(_resolve_profile_symbols(profile))
|
|
return symbols, f"profile:{profile}"
|
|
except Exception as exc:
|
|
fallback_file = PROFILE_SYMBOL_FALLBACK_FILES.get(profile)
|
|
if not fallback_file:
|
|
raise
|
|
path = Path(fallback_file)
|
|
if not path.exists():
|
|
raise
|
|
print(
|
|
f"[warn] profile resolution failed for {profile}: {exc}. "
|
|
f"Falling back to {path}.",
|
|
file=sys.stderr,
|
|
)
|
|
return _load_symbols_from_yaml(path), f"profile:{profile}:fallback_yaml:{path}"
|
|
|
|
|
|
def _trading_days_scanned(
|
|
bars_by_symbol: dict[str, dict[dt.date, dict[str, Any]]],
|
|
*,
|
|
start_date: dt.date,
|
|
end_date: dt.date,
|
|
) -> int:
|
|
spy_bars = bars_by_symbol.get("SPY") or {}
|
|
if spy_bars:
|
|
return sum(1 for date in spy_bars if start_date <= date <= end_date)
|
|
all_dates = {
|
|
date
|
|
for date_bars in bars_by_symbol.values()
|
|
for date in date_bars
|
|
if start_date <= date <= end_date
|
|
}
|
|
return len(all_dates)
|
|
|
|
|
|
def _write_trades_csv(path: Path, trades: list[DailyGainerTrade]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
rows = []
|
|
for trade in trades:
|
|
row = asdict(trade)
|
|
row["signal_date"] = trade.signal_date.isoformat()
|
|
row["entry_date"] = trade.entry_date.isoformat()
|
|
rows.append(row)
|
|
with path.open("w", newline="") as handle:
|
|
writer = csv.DictWriter(handle, fieldnames=list(rows[0].keys()) if rows else [])
|
|
if rows:
|
|
writer.writeheader()
|
|
writer.writerows(rows)
|
|
|
|
|
|
def _build_summary(
|
|
*,
|
|
trades: list[DailyGainerTrade],
|
|
universe_size: int,
|
|
universe_source: str,
|
|
start_date: dt.date,
|
|
end_date: dt.date,
|
|
criteria: DailyGainerCriteria,
|
|
trading_days_scanned: int,
|
|
) -> dict[str, Any]:
|
|
signal_dates = sorted({trade.signal_date for trade in trades})
|
|
by_rank: list[dict[str, Any]] = []
|
|
for rank in sorted({trade.rank for trade in trades}):
|
|
rank_trades = [trade for trade in trades if trade.rank == rank]
|
|
by_rank.append(_summarize_trade_bucket(rank_trades, label=f"rank_{rank}"))
|
|
|
|
return {
|
|
"metadata": {
|
|
"start_date": start_date.isoformat(),
|
|
"end_date": end_date.isoformat(),
|
|
"universe_size": universe_size,
|
|
"universe_source": universe_source,
|
|
"trading_days_scanned": trading_days_scanned,
|
|
"signal_days": len(signal_dates),
|
|
"criteria": {
|
|
"min_change_pct": round(criteria.min_change * 100.0, 2),
|
|
"max_change_pct": (
|
|
round(criteria.max_change * 100.0, 2)
|
|
if criteria.max_change is not None
|
|
else None
|
|
),
|
|
"min_volume_ratio": criteria.min_volume_ratio,
|
|
"min_close_location": criteria.min_close_location,
|
|
"min_avg_dollar_volume": criteria.min_avg_dollar_volume,
|
|
"max_gap_pct": (
|
|
round(criteria.max_gap * 100.0, 2)
|
|
if criteria.max_gap is not None
|
|
else None
|
|
),
|
|
"top_n": criteria.top_n,
|
|
},
|
|
"warning": (
|
|
"This is a reconstructed Yahoo-style probe from historical daily bars, "
|
|
"not an exact historical Yahoo page replay. Hold 1d/3d/5d returns are "
|
|
"measured from next-open entry to the close of the 1st/3rd/5th trading "
|
|
"day including the entry day."
|
|
),
|
|
},
|
|
"overall": _summarize_trade_bucket(trades, label="overall"),
|
|
"by_rank": by_rank,
|
|
}
|
|
|
|
|
|
def _summarize_trade_bucket(trades: list[DailyGainerTrade], *, label: str) -> dict[str, Any]:
|
|
signal_returns = [trade.signal_return for trade in trades]
|
|
entry_gaps = [trade.entry_gap_pct for trade in trades]
|
|
hold_1d = [trade.hold_1d_return for trade in trades if trade.hold_1d_return is not None]
|
|
hold_3d = [trade.hold_3d_return for trade in trades if trade.hold_3d_return is not None]
|
|
hold_5d = [trade.hold_5d_return for trade in trades if trade.hold_5d_return is not None]
|
|
return {
|
|
"label": label,
|
|
"trade_count": len(trades),
|
|
"signal_return": summarize_metric(signal_returns),
|
|
"entry_gap_pct": summarize_metric(entry_gaps),
|
|
"hold_1d_return": summarize_metric(hold_1d),
|
|
"hold_3d_return": summarize_metric(hold_3d),
|
|
"hold_5d_return": summarize_metric(hold_5d),
|
|
}
|
|
|
|
|
|
def _print_summary(summary: dict[str, Any]) -> None:
|
|
meta = summary["metadata"]
|
|
print("\n" + "=" * 100)
|
|
print("Yahoo-Style Daily Gainer Hypothesis Probe")
|
|
print("=" * 100)
|
|
print(
|
|
f"Window: {meta['start_date']} -> {meta['end_date']} "
|
|
f"Universe: {meta['universe_source']} ({meta['universe_size']} symbols)"
|
|
)
|
|
print(
|
|
f"Trading days scanned: {meta['trading_days_scanned']} "
|
|
f"Signal days: {meta['signal_days']}"
|
|
)
|
|
print(
|
|
"Criteria: "
|
|
f"change {meta['criteria']['min_change_pct']:.2f}%"
|
|
+ (
|
|
f" to {meta['criteria']['max_change_pct']:.2f}%"
|
|
if meta["criteria"]["max_change_pct"] is not None
|
|
else "+"
|
|
)
|
|
+ f", vol_ratio >= {meta['criteria']['min_volume_ratio']:.2f}, "
|
|
f"close_location >= {meta['criteria']['min_close_location']:.2f}, "
|
|
f"ADV >= ${meta['criteria']['min_avg_dollar_volume']:,.0f}, "
|
|
f"top_n = {meta['criteria']['top_n']}"
|
|
)
|
|
print(f"Note: {meta['warning']}")
|
|
|
|
header = (
|
|
f"\n{'Bucket':<10} {'Trades':>7} "
|
|
f"{'SigMean':>10} {'GapMean':>10} "
|
|
f"{'H1Mean':>10} {'H1WR':>8} "
|
|
f"{'H3Mean':>10} {'H3WR':>8} "
|
|
f"{'H5Mean':>10} {'H5WR':>8}"
|
|
)
|
|
print(header)
|
|
print("-" * len(header))
|
|
rows = [summary["overall"], *summary["by_rank"]]
|
|
for row in rows:
|
|
print(
|
|
f"{row['label']:<10} {row['trade_count']:>7} "
|
|
f"{_fmt_pct(row['signal_return']['mean']):>10} "
|
|
f"{_fmt_pct(row['entry_gap_pct']['mean']):>10} "
|
|
f"{_fmt_pct(row['hold_1d_return']['mean']):>10} "
|
|
f"{_fmt_wr(row['hold_1d_return']['win_rate']):>8} "
|
|
f"{_fmt_pct(row['hold_3d_return']['mean']):>10} "
|
|
f"{_fmt_wr(row['hold_3d_return']['win_rate']):>8} "
|
|
f"{_fmt_pct(row['hold_5d_return']['mean']):>10} "
|
|
f"{_fmt_wr(row['hold_5d_return']['win_rate']):>8}"
|
|
)
|
|
|
|
|
|
def _fmt_pct(value: float | None) -> str:
|
|
if value is None:
|
|
return "N/A"
|
|
return f"{value:+.2%}"
|
|
|
|
|
|
def _fmt_wr(value: float | None) -> str:
|
|
if value is None:
|
|
return "N/A"
|
|
return f"{value:.1%}"
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(
|
|
description="Reconstruct and probe Yahoo-style daily gainer continuation setups."
|
|
)
|
|
parser.add_argument("--start", help="Start date (YYYY, YYYY-MM, YYYY-MM-DD).")
|
|
parser.add_argument("--end", help="End date (YYYY, YYYY-MM, YYYY-MM-DD).")
|
|
parser.add_argument(
|
|
"--universe-profile",
|
|
choices=sorted(UNIVERSE_PROFILES),
|
|
default=None,
|
|
help="Resolve symbols from a current screener universe profile.",
|
|
)
|
|
parser.add_argument(
|
|
"--symbols-file",
|
|
help="YAML symbols file. If omitted, universe-profile is used.",
|
|
)
|
|
parser.add_argument(
|
|
"--symbol",
|
|
action="append",
|
|
default=[],
|
|
help="Explicit symbol override (repeatable).",
|
|
)
|
|
parser.add_argument(
|
|
"--symbol-cap",
|
|
type=int,
|
|
default=None,
|
|
help="Optional cap on the number of symbols for faster iteration.",
|
|
)
|
|
parser.add_argument("--min-change-pct", type=float, default=8.0)
|
|
parser.add_argument("--max-change-pct", type=float, default=25.0)
|
|
parser.add_argument("--min-volume-ratio", type=float, default=1.5)
|
|
parser.add_argument("--min-close-location", type=float, default=0.60)
|
|
parser.add_argument("--min-adv", type=float, default=50_000_000.0)
|
|
parser.add_argument(
|
|
"--max-gap-pct",
|
|
type=float,
|
|
default=None,
|
|
help="Optional max signal-day opening gap vs prior close.",
|
|
)
|
|
parser.add_argument("--top-n", type=int, default=3)
|
|
parser.add_argument(
|
|
"--output-dir",
|
|
default="data/analysis/yahoo_gainers_hypothesis_probe",
|
|
help="Directory for summary.json and trades.csv.",
|
|
)
|
|
parser.add_argument(
|
|
"--json",
|
|
action="store_true",
|
|
help="Print summary JSON in addition to writing files.",
|
|
)
|
|
parser.add_argument("--log-level", default="WARNING")
|
|
args = parser.parse_args()
|
|
|
|
configure_logging(args.log_level)
|
|
|
|
end_date = _parse_date(args.end, is_end=True) if args.end else _default_end_date()
|
|
start_date = _parse_date(args.start) if args.start else _default_start_date(end_date)
|
|
if start_date > end_date:
|
|
raise ValueError(f"start_date {start_date} is after end_date {end_date}")
|
|
|
|
symbols, universe_source = _resolve_symbols(args)
|
|
if args.symbol_cap is not None:
|
|
symbols = symbols[: max(0, args.symbol_cap)]
|
|
universe_source = f"{universe_source}[0:{max(0, args.symbol_cap)}]"
|
|
if not symbols:
|
|
raise ValueError("No symbols resolved for the probe.")
|
|
|
|
criteria = DailyGainerCriteria(
|
|
min_change=args.min_change_pct / 100.0,
|
|
max_change=(
|
|
None if args.max_change_pct is None else args.max_change_pct / 100.0
|
|
),
|
|
min_volume_ratio=args.min_volume_ratio,
|
|
min_close_location=args.min_close_location,
|
|
min_avg_dollar_volume=args.min_adv,
|
|
max_gap=None if args.max_gap_pct is None else args.max_gap_pct / 100.0,
|
|
top_n=max(1, args.top_n),
|
|
)
|
|
|
|
settings = get_settings()
|
|
fetch_symbols = sorted(set(symbols) | {"SPY"})
|
|
bars_by_symbol, _ = asyncio.run(
|
|
SnapshotStore._fetch_price_data(
|
|
fetch_symbols,
|
|
(start_date, end_date),
|
|
settings.stock_oracle_url,
|
|
)
|
|
)
|
|
probe_bars = {symbol: bars_by_symbol.get(symbol, {}) for symbol in symbols}
|
|
|
|
signals_by_date = collect_daily_gainer_signals(
|
|
bars_by_symbol=probe_bars,
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
criteria=criteria,
|
|
)
|
|
trades = evaluate_daily_gainer_signals(
|
|
signals_by_date=signals_by_date,
|
|
bars_by_symbol=probe_bars,
|
|
)
|
|
summary = _build_summary(
|
|
trades=trades,
|
|
universe_size=len(symbols),
|
|
universe_source=universe_source,
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
criteria=criteria,
|
|
trading_days_scanned=_trading_days_scanned(
|
|
bars_by_symbol,
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
),
|
|
)
|
|
|
|
output_dir = Path(args.output_dir)
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
summary_path = output_dir / "summary.json"
|
|
trades_path = output_dir / "trades.csv"
|
|
summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2))
|
|
_write_trades_csv(trades_path, trades)
|
|
|
|
_print_summary(summary)
|
|
print(f"\nSummary written to: {summary_path}")
|
|
print(f"Trades written to: {trades_path}")
|
|
if args.json:
|
|
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|