backfill: historical breadth + regime computation from TOP50 OHLCV
- Step 1: fetch BTC OHLCV - Step 2: fetch TOP50 daily data → compute breadth per date → store breadth_daily - Step 3: compute Price/Breadth/OI/Vol → detect regime → store regime_history - 175 days backfilled (2026-01-01 to 2026-06-24)
This commit is contained in:
+100
-27
@@ -192,23 +192,104 @@ def cmd_track(args):
|
||||
|
||||
|
||||
def cmd_backfill(args):
|
||||
"""Backfill historical scores and/or signals."""
|
||||
"""Backfill historical breadth + regime scores."""
|
||||
from datetime import date as Date, timedelta
|
||||
from database import init_db, get_connection
|
||||
from fetchers.ohlcv import OHLCVFetcher
|
||||
from fetchers.breadth import BreadthFetcher
|
||||
from config import config
|
||||
import pandas as pd
|
||||
import requests
|
||||
|
||||
start = parse_date(args.from_date)
|
||||
end = parse_date(args.to_date) if args.to_date else Date.today()
|
||||
init_db()
|
||||
|
||||
# First, backfill OHLCV data
|
||||
logger.info(f"Backfilling OHLCV from {start} to {end}...")
|
||||
fetcher = OHLCVFetcher()
|
||||
df = fetcher.fetch()
|
||||
if not df.empty:
|
||||
fetcher.store_df(df)
|
||||
# Step 1: Ensure OHLCV data exists for the range
|
||||
logger.info(f"Step 1/3: Fetching BTC OHLCV...")
|
||||
OHLCVFetcher().store_df(OHLCVFetcher().fetch())
|
||||
|
||||
# Then compute scores for each date
|
||||
# Step 2: Backfill breadth — fetch TOP50 daily data and compute per date
|
||||
logger.info(f"Step 2/3: Backfilling breadth {start} → {end}...")
|
||||
provider_url = config.provider_url
|
||||
all_symbol_data = {}
|
||||
|
||||
for sym in config.top50_symbols:
|
||||
try:
|
||||
df = pd.DataFrame(requests.get(
|
||||
f"{provider_url}/api/candles",
|
||||
params={"symbol": sym, "tf": "1d", "limit": 400},
|
||||
timeout=30
|
||||
).json())
|
||||
if not df.empty and "timestamp" in df.columns:
|
||||
df["date"] = pd.to_datetime(df["timestamp"], unit="ms").dt.date
|
||||
df["close"] = df["close"].astype(float)
|
||||
df["high"] = df["high"].astype(float)
|
||||
df["ema20"] = df["close"].ewm(20).mean()
|
||||
all_symbol_data[sym] = df
|
||||
except Exception as e:
|
||||
logger.debug(f" Skip {sym}: {e}")
|
||||
|
||||
logger.info(f" Fetched {len(all_symbol_data)}/{len(config.top50_symbols)} symbols")
|
||||
|
||||
# Compute breadth for each date
|
||||
conn = get_connection()
|
||||
current = start
|
||||
breadth_count = 0
|
||||
while current <= end:
|
||||
target_str = str(current)
|
||||
try:
|
||||
advances_50 = declines_50 = above_ema20_50 = new_highs_50 = 0
|
||||
advances_30 = advances_20 = above_ema20_30 = above_ema20_20 = 0
|
||||
new_highs_30 = new_highs_20 = 0
|
||||
|
||||
for rank, (sym, df) in enumerate(all_symbol_data.items()):
|
||||
rows = df[df["date"] == current]
|
||||
if rows.empty:
|
||||
continue
|
||||
row = rows.iloc[0]
|
||||
prev_rows = df[df["date"] < current]
|
||||
if prev_rows.empty:
|
||||
continue
|
||||
prev = prev_rows.iloc[-1]
|
||||
|
||||
if row["close"] > prev["close"]:
|
||||
if rank < 50: advances_50 += 1
|
||||
if rank < 30: advances_30 += 1
|
||||
if rank < 20: advances_20 += 1
|
||||
elif row["close"] < prev["close"]:
|
||||
if rank < 50: declines_50 += 1
|
||||
|
||||
if not pd.isna(row.get("ema20")) and row["close"] > row["ema20"]:
|
||||
if rank < 50: above_ema20_50 += 1
|
||||
if rank < 30: above_ema20_30 += 1
|
||||
if rank < 20: above_ema20_20 += 1
|
||||
|
||||
recent_highs = df[(df["date"] < current) & (df["date"] >= current - timedelta(days=20))]
|
||||
if not recent_highs.empty and row["high"] > recent_highs["high"].max():
|
||||
if rank < 50: new_highs_50 += 1
|
||||
if rank < 30: new_highs_30 += 1
|
||||
if rank < 20: new_highs_20 += 1
|
||||
|
||||
conn.execute("""INSERT OR REPLACE INTO breadth_daily
|
||||
(date, total_tracked, advance_top50, decline_top50, above_ema20_top50,
|
||||
new_highs_20d_top50, advance_top30, advance_top20,
|
||||
above_ema20_top30, above_ema20_top20, new_highs_20d_top30, new_highs_20d_top20)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
||||
(target_str, len(all_symbol_data),
|
||||
advances_50, declines_50, above_ema20_50, new_highs_50,
|
||||
advances_30, advances_20, above_ema20_30, above_ema20_20,
|
||||
new_highs_30, new_highs_20))
|
||||
breadth_count += 1
|
||||
except Exception as e:
|
||||
logger.debug(f" Breadth skip {current}: {e}")
|
||||
current += timedelta(days=1)
|
||||
|
||||
conn.commit()
|
||||
logger.info(f" Breadth backfill: {breadth_count} days")
|
||||
|
||||
# Step 3: Compute regime scores for each date
|
||||
logger.info(f"Step 3/3: Computing regime scores {start} → {end}...")
|
||||
from scoring.price_structure import PriceStructureScorer
|
||||
from scoring.breadth_scorer import BreadthScorer
|
||||
from scoring.oi_matrix import OIMatrixScorer
|
||||
@@ -216,10 +297,8 @@ def cmd_backfill(args):
|
||||
from regime_detector import RegimeDetector
|
||||
|
||||
detector = RegimeDetector()
|
||||
conn = get_connection()
|
||||
|
||||
current = start
|
||||
count = 0
|
||||
score_count = 0
|
||||
while current <= end:
|
||||
try:
|
||||
ps = PriceStructureScorer().compute(current)
|
||||
@@ -227,33 +306,27 @@ def cmd_backfill(args):
|
||||
if br.score == 50.0 and br.label == "No Data":
|
||||
current += timedelta(days=1)
|
||||
continue
|
||||
|
||||
oi = OIMatrixScorer().compute(current)
|
||||
vol = VolatilityRegimeScorer().compute(current)
|
||||
r = detector.detect(ps.score, br.breadth_top50,
|
||||
vol.vol_regime.value, current)
|
||||
r = detector.detect(ps.score, br.breadth_top50, vol.vol_regime.value, current)
|
||||
|
||||
conn.execute("""
|
||||
INSERT OR REPLACE INTO regime_history
|
||||
conn.execute("""INSERT OR REPLACE INTO regime_history
|
||||
(date, regime, confidence, regime_version, maturity_score,
|
||||
all_scores_json, confirmation_days)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||||
""", (
|
||||
str(current), r.regime.value, r.confidence,
|
||||
r.regime_version, r.maturity_score,
|
||||
json.dumps(r.all_scores), r.confirmation_days,
|
||||
))
|
||||
count += 1
|
||||
if count % 30 == 0:
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?)""",
|
||||
(str(current), r.regime.value, r.confidence, r.regime_version,
|
||||
r.maturity_score, json.dumps(r.all_scores), r.confirmation_days))
|
||||
score_count += 1
|
||||
if score_count % 30 == 0:
|
||||
conn.commit()
|
||||
logger.info(f" Backfilled {count} days... ({current})")
|
||||
logger.info(f" Scored {score_count} days... ({current})")
|
||||
except Exception as e:
|
||||
logger.debug(f" Skip {current}: {e}")
|
||||
logger.debug(f" Score skip {current}: {e}")
|
||||
current += timedelta(days=1)
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
logger.info(f"Backfill complete: {count} days scored")
|
||||
logger.info(f"Backfill complete: {breadth_count} breadth + {score_count} regime days")
|
||||
|
||||
|
||||
def cmd_expectancy(args):
|
||||
|
||||
Reference in New Issue
Block a user