Spaces:
Running
Running
| #!/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() | |