momentum devandClaude Fable 5
af8137dfd7
pg_schema.sql: drop CREATE TABLE for 10 zero-reference tables (operator DROP list in plans/002-deletions.log — production tables untouched until run). projections.py: remove 4 zero-consumer query functions. Clear all 'plan 002' ruff pins from the CI gate (files now F-clean); api/main.py:F821 pin remains for plan 007. README repo-map/test-count fixed; pydantic-settings removed (zero imports). Full receipts: plans/002-deletions.log Suite: 351 passed (16 schwab/dead-module tests removed), 15 failures = the known WSL-loopback dashboard set only. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
240 lines
11 KiB
Python
240 lines
11 KiB
Python
"""Year-long gapper-history reconstruction — statistical power for the core edge.
|
|
|
|
Every conclusion so far (including the LIVE confirmation gate) rests on 19-26 days of history.
|
|
The paid plan's 9+ years of SIP data lets us fix that surgically instead of ingesting the whole
|
|
market at minute resolution:
|
|
|
|
1. daily — pull ~1 year of DAILY bars (SIP) for the full listed US-equity universe
|
|
(~10-11k symbols; ~2.7M rows; cheap).
|
|
2. gappers — reconstruct each historical day's TOP GAPPERS from those dailies
|
|
(open/prev_close-1 >= 10%, open $1-20, day volume >= 300k; top 20/day)
|
|
into the `historical_gappers` table (~250 days x ~20 = ~4-5k symbol-days).
|
|
3. minutes — pull SIP minute bars (04:00-20:00 ET, premarket included) for JUST those
|
|
gapper symbol-days (~2-4M rows). Idempotent; skips symbol-days already present.
|
|
4. analyze — replay the ORB through the year: OR(5-min) trigger, first-cross minute,
|
|
pessimistic bracket outcome; report conversion BY CONFIRMATION-MINUTE bucket
|
|
and by month. PRE-REGISTERED BAR (locked before the data): the confirmation
|
|
gate GENERALIZES if the early(<=15min)-cross bucket beats the late bucket by
|
|
>= +0.25R expectancy AND is the best bucket in >= 70% of months.
|
|
|
|
python -m research.gapper_history all # chain 1-4
|
|
python -m research.gapper_history <step> # individual steps (resumable)
|
|
"""
|
|
|
|
import argparse
|
|
import time
|
|
from collections import defaultdict
|
|
from datetime import datetime, timedelta
|
|
|
|
import os
|
|
import pandas as pd
|
|
from dotenv import load_dotenv
|
|
load_dotenv(os.path.join(os.path.dirname(__file__), "..", ".env"))
|
|
|
|
from alpaca_paper.client import AlpacaPaperClient # noqa: E402
|
|
from alpaca_paper.settings import AlpacaPaperSettings # noqa: E402
|
|
from research.ingestion.market_data import ( # noqa: E402
|
|
upsert_daily_bars, upsert_minute_bars)
|
|
from research.multi_schema import open_research_db # noqa: E402
|
|
|
|
LOOKBACK_DAYS = 380 # ~1 year of trading days + padding
|
|
GAP_MIN = 0.10
|
|
PRICE_LO, PRICE_HI = 1.0, 20.0
|
|
DAY_VOL_MIN = 300_000.0
|
|
TOP_N = 20
|
|
ADVERSE = 1.002 # +0.2% adverse fill in the replay
|
|
CONFIRM_BUCKETS = [(0, 15, "00-15m"), (15, 30, "15-30m"), (30, 60, "30-60m"),
|
|
(60, 120, "60-120m"), (120, 999, "120m+")]
|
|
|
|
|
|
def _client():
|
|
return AlpacaPaperClient(AlpacaPaperSettings.from_env())
|
|
|
|
|
|
def _universe(client) -> list[str]:
|
|
"""Listed US equities, junk filtered (no OTC per assets API; drop units/warrants)."""
|
|
out = []
|
|
for a in client.get_assets():
|
|
sym = (a.get("symbol") or "").strip()
|
|
if not sym or "." in sym or "/" in sym or len(sym) > 5:
|
|
continue
|
|
if len(sym) == 5 and sym[-1] in ("W", "R", "U"): # warrants/rights/units
|
|
continue
|
|
if a.get("exchange") == "OTC":
|
|
continue
|
|
out.append(sym)
|
|
return sorted(set(out))
|
|
|
|
|
|
def step_daily() -> None:
|
|
con = open_research_db("market")
|
|
client = _client()
|
|
syms = _universe(client)
|
|
start = (datetime.utcnow() - timedelta(days=LOOKBACK_DAYS)).strftime("%Y-%m-%dT00:00:00Z")
|
|
print(f"universe: {len(syms)} symbols; pulling dailies since {start[:10]}", flush=True)
|
|
t0, rows = time.time(), 0
|
|
for i in range(0, len(syms), 100):
|
|
chunk = syms[i:i + 100]
|
|
try:
|
|
by_symbol = client.get_daily_bars(chunk, start_iso=start, feed="sip")
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f" ! chunk {i}: {str(exc)[:70]}", flush=True)
|
|
time.sleep(2)
|
|
continue
|
|
for sym, bars in by_symbol.items():
|
|
if bars:
|
|
rows += upsert_daily_bars(con, sym, bars)
|
|
if (i // 100) % 10 == 0:
|
|
print(f" ... {i + len(chunk)}/{len(syms)} symbols, {rows} rows ({time.time()-t0:.0f}s)", flush=True)
|
|
print(f"daily done: {rows} rows in {time.time()-t0:.0f}s", flush=True)
|
|
|
|
|
|
def step_gappers() -> None:
|
|
con = open_research_db("market")
|
|
con.execute("""CREATE TABLE IF NOT EXISTS historical_gappers (
|
|
session_date DATE NOT NULL, symbol TEXT NOT NULL,
|
|
gap_pct DOUBLE PRECISION, day_volume DOUBLE PRECISION, rank INT,
|
|
PRIMARY KEY (session_date, symbol))""")
|
|
rows = con.execute(
|
|
"SELECT symbol, trade_date, open, previous_close, volume FROM daily_bars "
|
|
"WHERE previous_close > 0 AND open BETWEEN ? AND ? AND volume >= ? "
|
|
"AND (open / previous_close - 1.0) >= ?",
|
|
[PRICE_LO, PRICE_HI, DAY_VOL_MIN, GAP_MIN]).fetchall()
|
|
by_day = defaultdict(list)
|
|
for sym, d, o, pc, v in rows:
|
|
by_day[d].append((sym, float(o) / float(pc) - 1.0, float(v)))
|
|
n = 0
|
|
for d, cands in by_day.items():
|
|
cands.sort(key=lambda x: -x[1])
|
|
for rank, (sym, gap, vol) in enumerate(cands[:TOP_N], 1):
|
|
con.execute(
|
|
"INSERT INTO historical_gappers (session_date, symbol, gap_pct, day_volume, rank) "
|
|
"VALUES (?, ?, ?, ?, ?) ON CONFLICT (session_date, symbol) DO UPDATE SET "
|
|
"gap_pct=EXCLUDED.gap_pct, day_volume=EXCLUDED.day_volume, rank=EXCLUDED.rank",
|
|
[d, sym, gap, vol, rank])
|
|
n += 1
|
|
print(f"gappers: {n} symbol-days across {len(by_day)} sessions", flush=True)
|
|
|
|
|
|
def step_minutes() -> None:
|
|
con = open_research_db("market")
|
|
client = _client()
|
|
todo = con.execute("""
|
|
SELECT hg.session_date, hg.symbol FROM historical_gappers hg
|
|
WHERE NOT EXISTS (SELECT 1 FROM minute_bars mb WHERE mb.symbol = hg.symbol
|
|
AND mb.session_date = hg.session_date
|
|
GROUP BY mb.symbol HAVING count(*) >= 50)
|
|
ORDER BY hg.session_date""").fetchall()
|
|
by_day = defaultdict(list)
|
|
for d, sym in todo:
|
|
by_day[d].append(sym)
|
|
print(f"minutes: {len(todo)} symbol-days to fetch across {len(by_day)} sessions", flush=True)
|
|
t0, rows = time.time(), 0
|
|
for j, (d, syms) in enumerate(sorted(by_day.items())):
|
|
start = f"{d.isoformat()}T08:00:00Z"
|
|
end = f"{(d + timedelta(days=1)).isoformat()}T00:00:00Z"
|
|
try:
|
|
by_symbol = client.get_minute_bars(syms, start_iso=start, end_iso=end, feed="sip")
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f" ! {d}: {str(exc)[:70]}", flush=True)
|
|
time.sleep(2)
|
|
continue
|
|
for sym, bars in by_symbol.items():
|
|
if bars:
|
|
rows += upsert_minute_bars(con, sym, bars, source_provider="alpaca_sip")
|
|
if j % 20 == 0:
|
|
print(f" ... {j}/{len(by_day)} sessions, {rows} rows ({time.time()-t0:.0f}s)", flush=True)
|
|
print(f"minutes done: {rows} rows in {time.time()-t0:.0f}s", flush=True)
|
|
|
|
|
|
def step_analyze() -> None:
|
|
con = open_research_db("market")
|
|
days = [r[0] for r in con.execute(
|
|
"SELECT DISTINCT session_date FROM historical_gappers ORDER BY session_date").fetchall()]
|
|
results = []
|
|
t0 = time.time()
|
|
for d in days:
|
|
syms = [r[0] for r in con.execute(
|
|
"SELECT symbol FROM historical_gappers WHERE session_date=?", [d]).fetchall()]
|
|
if not syms:
|
|
continue
|
|
ph = ",".join(["?"] * len(syms))
|
|
b = pd.DataFrame(con.execute(
|
|
f"SELECT symbol, timestamp, open, high, low, close FROM minute_bars "
|
|
f"WHERE session_date=? AND is_regular_hours AND symbol IN ({ph}) "
|
|
f"ORDER BY symbol, timestamp", [d] + syms).fetchall(),
|
|
columns=["symbol", "ts", "open", "high", "low", "close"])
|
|
for col in ("open", "high", "low", "close"):
|
|
b[col] = b[col].astype(float)
|
|
for sym, bb in b.groupby("symbol"):
|
|
bb = bb.reset_index(drop=True)
|
|
if len(bb) < 20:
|
|
continue
|
|
orh = float(bb["high"].iloc[:5].max())
|
|
orl = float(bb["low"].iloc[:5].min())
|
|
if orh <= orl:
|
|
continue
|
|
fwd = bb.iloc[5:].reset_index(drop=True)
|
|
cross = fwd[fwd["high"] >= orh]
|
|
if cross.empty:
|
|
continue
|
|
ci = int(cross.index[0])
|
|
confirm_min = ci + 5 # minutes after the open
|
|
fill = orh * ADVERSE
|
|
R = fill - orl
|
|
if R <= 0:
|
|
continue
|
|
out_r = None
|
|
for _, row in fwd.iloc[ci + 1:].iterrows():
|
|
if row["low"] <= orl:
|
|
out_r = -1.0
|
|
break
|
|
if row["high"] >= fill + 2 * R:
|
|
out_r = 2.0
|
|
break
|
|
if out_r is None:
|
|
out_r = (float(fwd["close"].iloc[-1]) - fill) / R
|
|
results.append(dict(month=str(d)[:7], confirm=confirm_min, r=out_r))
|
|
df = pd.DataFrame(results)
|
|
print(f"\n=== YEAR-LONG ORB REPLAY (n={len(df)} gapper symbol-days, {time.time()-t0:.0f}s) ===")
|
|
print("\nBY CONFIRMATION MINUTE (the gate's core claim):")
|
|
for lo, hi, lab in CONFIRM_BUCKETS:
|
|
sub = df[(df["confirm"] > lo) & (df["confirm"] <= hi)]
|
|
if len(sub):
|
|
print(f" {lab:9} n={len(sub):5} expR={sub['r'].mean():+.3f} "
|
|
f"win%={100*(sub['r']>0).mean():.0f} medR={sub['r'].median():+.2f}")
|
|
early = df[df["confirm"] <= 15]
|
|
late = df[df["confirm"] > 15]
|
|
if len(early) and len(late):
|
|
delta = early["r"].mean() - late["r"].mean()
|
|
print(f"\nEARLY(<=15m) expR {early['r'].mean():+.3f} vs LATE {late['r'].mean():+.3f} "
|
|
f"-> delta {delta:+.3f}R [{'HOLDS' if delta >= 0.25 else 'FAILS'} the +0.25R bar]")
|
|
# month-by-month consistency: is early the best bucket most months?
|
|
wins = 0
|
|
months = sorted(df["month"].unique())
|
|
for m in months:
|
|
dm = df[df["month"] == m]
|
|
e, l = dm[dm["confirm"] <= 15], dm[dm["confirm"] > 15]
|
|
if len(e) >= 10 and len(l) >= 10 and e["r"].mean() > l["r"].mean():
|
|
wins += 1
|
|
print(f"month-by-month: early beats late in {wins}/{len(months)} months "
|
|
f"[{'HOLDS' if wins >= 0.7 * len(months) else 'FAILS'} the 70% consistency bar]")
|
|
|
|
|
|
def main() -> None:
|
|
ap = argparse.ArgumentParser(description="year-long gapper history reconstruction")
|
|
ap.add_argument("step", choices=["daily", "gappers", "minutes", "analyze", "all"])
|
|
args = ap.parse_args()
|
|
steps = {"daily": step_daily, "gappers": step_gappers,
|
|
"minutes": step_minutes, "analyze": step_analyze}
|
|
if args.step == "all":
|
|
for name in ("daily", "gappers", "minutes", "analyze"):
|
|
print(f"\n########## {name} ##########", flush=True)
|
|
steps[name]()
|
|
else:
|
|
steps[args.step]()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|