Blame for engine/export_bundle.py

#!/usr/bin/env python3
"""Slipstream — data-bundle exporter (runs LOCALLY, where the Sharadar cache lives).

Rule (v2, Oct 2026 — validated in data/research/slipstream/quality, Studies 4–8):
  managers  every 13F filer that is a concentrated stock-picker that quarter (10–150 positions,
            $250M–$100B, top-10 ≥ 40%, filed the prior quarter) — built by
            scripts/slipstream/manager_universe.py --sets CONC (no curated list, no insiders)
  signal    consensus = 2·new + add − 2·exit conviction positions; eligible if > 0; size = 13F $
  quality   ROA (SF1 ART, latest filing public by the signal date) in the top 30% of its sector,
            ranked across the month's top-1,000 universe

This script exports a small, self-contained bundle the HF Space computes the fund from:

  data/candidates.parquet  — latest eligible names (top-1000 & ROA-30 & consensus BUY),
                             with score, $-size and sector
  data/prices.parquet      — daily adj close for the top-1000 universe + SPY, last ~400
                             trading days (drives regime, momentum and sector-RS on HF)
  data/meta.json           — as-of date, universe, generated-at

13F is quarterly and ~45d lagged, so re-running this quarterly (and committing the bundle
to the Space) keeps the fund current. Run:  python3 engine/export_bundle.py
"""
from __future__ import annotations

import json
import sys
from datetime import datetime, timezone
from pathlib import Path

import pandas as pd

import os as _os
# env-overridable: this runs inside the Refinery container too, where the Quant Model
# code lives at /app/quantmodel rather than on the Mac.
QM = Path(_os.environ.get("QM_DIR", "/Users/sttp/Desktop/Quant Model"))
sys.path.insert(0, str(QM))
import scripts.follow_the_money_backtest as B  # noqa: E402

DATA = Path(_os.environ.get("SLIP_BUNDLE_DIR", Path(__file__).resolve().parent.parent / "data"))
LOOKBACK_DAYS = 400   # ≥ 252 for momentum / 200 for regime / 252 for sector-RS
# the concentrated-manager panel written by manager_universe.py (refresh_everything step 7b)
CONC_DIR = Path(_os.environ.get("SLIP_CONC_DIR", QM / "data/research/conc/panels/CONC"))
PULL = Path(_os.environ.get("SLIP_PULL_DIR", QM / "data/raw/ftm_pull"))
ROA_KEEP = 0.30                       # top 30% of each sector by ROA
# parity testing only: restrict to signal dates before this day / prices through this day
MAX_DATE = _os.environ.get("SLIP_EXPORT_BEFORE")
PRICES_THROUGH = _os.environ.get("SLIP_PRICES_THROUGH")


def consensus_frame(rows: pd.DataFrame) -> pd.DataFrame:
    """2·new + add − 2·exit among concentrated managers (scripts/slipstream/backtest.signal_frame 'consensus')."""
    df = rows.copy()
    col = lambda c: df[c].fillna(0.0) if c in df.columns else pd.Series(0.0, index=df.index)
    df["score"] = 2 * col("guru_new_buys") + col("guru_adds") - 2 * col("guru_exits")
    df["ok"] = df["score"] > 0
    df["size"] = col("guru_conv_value").clip(lower=0)
    return df


def roa_gate(d: pd.Timestamp, universe: list, sector: dict) -> set:
    """Names whose latest public ROA is in the top ROA_KEEP of their sector across `universe`
    (scripts/slipstream/backtest.Data.roa_pct / quality_set 'roa_sector')."""
    import duckdb
    f = duckdb.connect().execute(f"""
        SELECT ticker, CAST(datekey AS DATE) AS dk, TRY_CAST(roa AS DOUBLE) AS roa
        FROM read_csv_auto('{PULL / "SF1.csv"}', header=true, all_varchar=true)
        WHERE dimension = 'ART' AND datekey IS NOT NULL AND TRY_CAST(roa AS DOUBLE) IS NOT NULL
          AND CAST(datekey AS DATE) >= DATE '2012-06-01' AND CAST(datekey AS DATE) <= DATE '{d.date()}'""").df()
    f["dk"] = pd.to_datetime(f["dk"])
    f = f[f["ticker"].isin(universe)].sort_values("dk")
    x = f.groupby("ticker")["roa"].last().to_frame()
    x["sector"] = [sector.get(t, "?") for t in x.index]
    x["pct"] = x.groupby("sector")["roa"].rank(pct=True)
    return set(x.index[x["pct"] > 1 - ROA_KEEP])


def main():
    DATA.mkdir(parents=True, exist_ok=True)
    B.PANEL, B.PRICES = CONC_DIR / "ftm_panel.parquet", CONC_DIR / "ftm_prices.parquet"
    B.DAILY, B.SF1, B.TICKERS = PULL / "DAILY.csv", PULL / "SF1.csv", PULL / "TICKERS.csv"
    if not B.PANEL.exists():
        raise SystemExit(f"missing {B.PANEL} — run scripts/slipstream/manager_universe.py --sets CONC first")
    base_panel = QM / "data/research/ftm_panel.parquet"
    if not _os.environ.get("SLIP_CONC_DIR") and base_panel.exists() and B.PANEL.stat().st_mtime < base_panel.stat().st_mtime:
        # the concentrated-manager panel must be rebuilt AFTER the base panel in the same run,
        # or the bundle would ship last week's 13F signals as if fresh
        raise SystemExit(f"stale {B.PANEL}: older than {base_panel} — the conc_panel step did not run")
    panel, R, spy, sector, wide = B.load()
    if PRICES_THROUGH:
        wide = wide.loc[:PRICES_THROUGH]
        R = R.loc[:PRICES_THROUGH]
    valid = set(R.columns)
    bdr = {pd.Timestamp(d): g for d, g in panel.groupby("date")}
    # Use the LATEST signal rows in the panel (weekly cadence), not only monthly firsts.
    # The monthly-only pick made every live rebalance trade the prior month's signal row,
    # which — because 13F filings land mid-quarter (45-day deadline) — lagged the fresh
    # quarter by a whole cycle: on 2026-09-01 the fund would have traded the Aug-3 row
    # built on Q1 13F, while the Aug-17 weekly row already carried Q2. The backtest's PIT
    # semantics assume signals as-of the rebalance date, so freshest-usable matches the
    # validated rule more closely, not less. Only the last few dates are screened
    # (universe/quality checks hit the multi-GB DAILY/SF1 CSVs per date).
    dts = sorted(pd.to_datetime(panel["date"]).unique())
    if MAX_DATE:
        dts = [x for x in dts if pd.Timestamp(x) < pd.Timestamp(MAX_DATE)]
    dts = dts[-6:]
    uni = B.mktcap_universe(dts, 1000)

    # latest rebalance date with a usable book
    d = next((pd.Timestamp(c) for c in reversed(dts)
              if bdr.get(pd.Timestamp(c)) is not None and (valid & uni.get(pd.Timestamp(c), set()))), None)
    if d is None:
        raise SystemExit("no valid rebalance date")
    universe = sorted(uni.get(d, set()) & valid)
    qual = roa_gate(d, universe, sector)

    # candidates: eligible names with a consensus BUY among concentrated managers at d
    sf = consensus_frame(bdr[d])
    elig = valid & set(universe) & qual
    cand = sf[sf["ok"] & sf["ticker"].isin(elig)].copy()
    cand["sector"] = cand["ticker"].map(lambda t: sector.get(t, "?"))
    cand = cand[["ticker", "sector", "score", "size"]].reset_index(drop=True)
    cand.to_parquet(DATA / "candidates.parquet", index=False)

    # prices: universe + SPY, last LOOKBACK_DAYS trading days → long form
    cols = [t for t in (set(universe) | {"SPY"} | set(cand["ticker"])) if t in wide.columns]
    px = wide[cols].tail(LOOKBACK_DAYS)
    long = px.reset_index().melt(id_vars=px.index.name or "index", var_name="ticker",
                                 value_name="px").dropna()
    long.columns = ["date", "ticker", "px"]
    long.to_parquet(DATA / "prices.parquet", index=False)

    meta = {
        "fund": "Slipstream",
        "rule_version": "v2-2026-10 · CONC consensus · ROA top-30% in sector · 10% cap",
        "as_of": str(d.date()),
        "universe": universe,
        "sectors": {t: sector.get(t, "?") for t in cols},
        "n_candidates": int(len(cand)),
        "price_start": str(px.index.min().date()),
        "price_end": str(px.index.max().date()),
        "generated_at": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC"),
    }
    (DATA / "meta.json").write_text(json.dumps(meta, indent=1))
    print(f"→ bundle in {DATA}")
    print(f"  as_of {meta['as_of']} · {len(cand)} candidates · {len(universe)} universe · "
          f"prices {meta['price_start']}→{meta['price_end']} ({len(long):,} rows)")


if __name__ == "__main__":
    main()