From 2c19e5efe577627d6b29c0f2e1c89632a0f23372 Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Tue, 29 Sep 2026 01:03:50 +0800 Subject: [PATCH] feat: US 3-sleeve portfolio and lowrisk_top20; US portfolio research harness Register two live strategies from the 2026-09-29 US research: - lowrisk_top20: idiosyncratic vol + low beta, top 20, monthly, buffer 2 (new `low_risk` signal in strategies/robust_momentum.py) - us_three_sleeve: 1/3 each of overnight_recovery_top10, lowrisk_top20 and trend_rider_v7_vt24 in one account (strategies/sleeve_portfolio.py, a mixed stock+ETF strategy). Sleeve shares drift with their returns and reset at a month start once any is >25% off its share. Research (research/lab, S&P 500 PIT, next-open fills): - rotation.py / persistence.py: walk-forward factor weighting (factor momentum, rolling IC, market state) and last-year's-winner selection do not beat a fixed composite - us_sweep.py: holdings count, z/vol weights, dynamic N, tranching and market overlays (Faber SMA, de-risk vol target, statistical jump model, momentum state); engine.py gains zvol weighting and dyn_n - fetch_us_fundamentals.py / fundamentals.py / fund_eval.py: SEC EDGAR point-in-time fundamentals (first-filed values) and factor tests - trend_rider_long.py / us_final.py: 2010-2026 TrendRider curves and the 3-sleeve check (2011-2026 Sharpe 1.21, CAGR 21.9%, MaxDD -23%) Web-UI engine, 2017-09..2026-09, $2/order: us_three_sleeve CAGR 28.9%, Sharpe 1.49, MaxDD -24.8%. Co-Authored-By: Claude Opus 5.5 (1M context) --- CLAUDE.md | 10 +- bridge.py | 13 ++ research/lab/engine.py | 43 +++- research/lab/fetch_us_fundamentals.py | 184 ++++++++++++++++ research/lab/fund_eval.py | 120 +++++++++++ research/lab/fundamentals.py | 292 ++++++++++++++++++++++++++ research/lab/overlays.py | 169 +++++++++++++++ research/lab/persistence.py | 98 +++++++++ research/lab/rotation.py | 258 +++++++++++++++++++++++ research/lab/trend_rider_long.py | 75 +++++++ research/lab/us_final.py | 155 ++++++++++++++ research/lab/us_sweep.py | 184 ++++++++++++++++ strategies/robust_momentum.py | 21 ++ strategies/sleeve_portfolio.py | 140 ++++++++++++ tests/test_sleeve_portfolio.py | 93 ++++++++ trader.py | 9 + 16 files changed, 1857 insertions(+), 7 deletions(-) create mode 100644 research/lab/fetch_us_fundamentals.py create mode 100644 research/lab/fund_eval.py create mode 100644 research/lab/fundamentals.py create mode 100644 research/lab/persistence.py create mode 100644 research/lab/rotation.py create mode 100644 research/lab/trend_rider_long.py create mode 100644 research/lab/us_final.py create mode 100644 research/lab/us_sweep.py create mode 100644 strategies/sleeve_portfolio.py create mode 100644 tests/test_sleeve_portfolio.py diff --git a/CLAUDE.md b/CLAUDE.md index b2141f7..310a91c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -47,6 +47,11 @@ uv run python -m research.lab.fetch_cn_yahoo # CSI 300/500 ever-member PYTHONPATH=. uv run python -m research.lab.factor_scan --market us # or cn_hs300 PYTHONPATH=. uv run python -m research.lab.stepwise --market us --objective maximin PYTHONPATH=. uv run python -m research.lab.evaluate --market us # candidates vs champions +uv run python -m research.lab.fetch_us_fundamentals # SEC EDGAR point-in-time facts + Yahoo splits (US) +PYTHONPATH=. uv run python -m research.lab.rotation --market us # walk-forward factor weighting schemes +PYTHONPATH=. uv run python -m research.lab.persistence # does last year's best combo persist? +PYTHONPATH=. uv run --with numba python -m research.lab.us_sweep --part overlays # construction / overlay sweeps +PYTHONPATH=. uv run --with numba python -m research.lab.us_final # 3-sleeve US portfolio check # Setup uv sync # Install/sync dependencies @@ -71,7 +76,7 @@ uv run python -m unittest discover tests **Backtest engine** (`main.py`): Orchestrates data loading, strategy execution, and visualization. `backtest()` has two modes. Close mode (no `open_data`, used by `main.py` and old research scripts) is vectorized and trades at the close the signal saw. Live mode (`open_data` given, used by the web UI through `backtest_runner.py`) keeps the book like `trader.py`: same signal timing (`strategy_positions`), holdings that drift, the $50/0.1% order floor, `trader.RESIZE_BAND`, fees on the orders placed (`simulate_live_book`). -**Daily trader** (`trader.py`): Live/forward-testing system with persistent portfolio state in `data/trader_{market}_{strategy}.json`. The `auto` subcommand runs both signal generation and execution in a single invocation — designed for cron. The `simulate` subcommand replays a date range day-by-day with realistic portfolio tracking (fractional shares, cash, commissions). Orders: whole A-share lots are allocated nearest-lot within cash (`_allocate_lots`); a held name is resized only when ≥ `RESIZE_BAND` (10%) off target. Available strategies: `recovery_mom_top10`, `recovery_mom_top20`, `momentum`, `momentum_quality`, `dual_momentum`, `inverse_vol`, `trend_following`, `buy_and_hold`, 32 factor-combo strategies (`fc_{signal}_{freq}` — see `strategies/factor_combo.py`), and the robust-momentum bar strategies `overnight_recovery_top10`, `lowrisk_momentum_top15`, `lowbeta_sharpe_mom_top20`. +**Daily trader** (`trader.py`): Live/forward-testing system with persistent portfolio state in `data/trader_{market}_{strategy}.json`. The `auto` subcommand runs both signal generation and execution in a single invocation — designed for cron. The `simulate` subcommand replays a date range day-by-day with realistic portfolio tracking (fractional shares, cash, commissions). Orders: whole A-share lots are allocated nearest-lot within cash (`_allocate_lots`); a held name is resized only when ≥ `RESIZE_BAND` (10%) off target. Available strategies: `recovery_mom_top10`, `recovery_mom_top20`, `momentum`, `momentum_quality`, `dual_momentum`, `inverse_vol`, `trend_following`, `buy_and_hold`, 32 factor-combo strategies (`fc_{signal}_{freq}` — see `strategies/factor_combo.py`), the robust-momentum bar strategies `overnight_recovery_top10`, `lowrisk_momentum_top15`, `lowbeta_sharpe_mom_top20`, `lowrisk_top20`, and the US sleeve portfolio `us_three_sleeve`. **Live books** (`trader.LIVE_BOOKS`): a book is a cohort of strategies, each with its own account, started on one date with one capital and a flat fee per order. The original book (every live strategy, `MONITOR_CAPITAL`, `MARKET_FEES`) keeps state in `data/trader_*.json`; a named book keeps it in `data/books//`. The monitor runs every book; `auto`/`status`/`log`/`compare`/`replay` and `bridge.py live` take `--book`, and `bridge.py books` lists them for the UI's book selector. `selected_2026_09` (29 strategies, US $10,000 / CN ¥100,000, $2 / ¥10 per order) starts on 2026-09-25 (US) and 2026-09-28 (CN). @@ -88,7 +93,8 @@ uv run python -m unittest discover tests - `momentum_quality.py` — Momentum + return consistency + low drawdown - `adaptive_momentum.py` — Momentum weighted by inverse volatility - `recovery_momentum.py` — Recovery (price/63d low) + 12-1mo momentum composite. Best US performer. -- `robust_momentum.py` — Bar strategies from the 2026-09 PIT research (report: https://claude.ai/artifact/Mo9nrzs5SWj5FfyshERyWS): US `overnight_recovery` (overnight-return momentum + distance from 126d low), CN `lowrisk_momentum` (amplitude-cut momentum + low beta + price-volume corr + momentum/vol) and close-only `lowbeta_sharpe_mom`. Top-N with rank hysteresis, rebalance every 10 days, weights drift between rebalances. +- `robust_momentum.py` — Bar strategies from the 2026-09 PIT research (report: https://claude.ai/artifact/Mo9nrzs5SWj5FfyshERyWS): US `overnight_recovery` (overnight-return momentum + distance from 126d low), CN `lowrisk_momentum` (amplitude-cut momentum + low beta + price-volume corr + momentum/vol) and close-only `lowbeta_sharpe_mom`. Top-N with rank hysteresis, rebalance every 10 days, weights drift between rebalances. `low_risk` (US `lowrisk_top20`: idiosyncratic vol + low beta, top 20, monthly) is the defensive sleeve of the 2026-09-29 US research (report: https://claude.ai/artifact/VGhZvKkS9UBYZkLeqGjcNC). +- `sleeve_portfolio.py` — Several strategies in one account with fixed capital shares. `us_three_sleeve` = 1/3 each of `overnight_recovery_top10`, `lowrisk_top20`, `trend_rider_v7_vt24` (a mixed stock+ETF strategy); sleeve shares drift with their returns and reset at a month start once any is >25% off. 2011-2026 research: Sharpe 1.21 vs 0.88-1.03 for the sleeves alone. Research finding behind it: choosing factors by their recent record (rotation, factor momentum, last year's winner) did not beat a fixed composite; construction knobs (N, weighting, dynamic N) barely matter. - `factor_combo.py` — Configurable factor-combination strategies with daily/weekly/biweekly/monthly rebalancing. US champions: `rec_mfilt+deep_upvol`, `ma200+mom7m+rec126`, `rec_mfilt+ma200`, `mom7m+rec126`. CN champions: `up_cap+quality_mom`, `down_resil+qual_mom`, `rec63+mom_gap`, `up_cap+mom_gap` — these were mined on today's CSI 300 members; on PIT members they earn ~3-7% a year (2016-2026). All registered in trader.py as `fc_{signal}_{freq}` (e.g., `fc_rec_mfilt_deep_upvol_monthly`). 32 new strategies total. **Metrics** (`metrics.py`): Standalone functions for portfolio analytics (Sharpe, Sortino, Calmar, max drawdown, etc.). `summary()` prints a formatted report and returns a dict. diff --git a/bridge.py b/bridge.py index f42e946..1bbe93c 100644 --- a/bridge.py +++ b/bridge.py @@ -453,6 +453,19 @@ STRATEGY_META = { "params": {}, "markets": ["us", "cn"], }, + # --- US portfolio research (2026-09-29) --- + "lowrisk_top20": { + "label": "Robust: Low Risk (IVol + Beta) Top 20", + "category": "factor_combo", + "params": {}, + "markets": ["us", "cn"], + }, + "us_three_sleeve": { + "label": "US 3-Sleeve: Momentum + Low Risk + Trend Rider (1/3 each)", + "category": "tactical_allocation", + "params": {}, + "markets": ["us"], + }, } diff --git a/research/lab/engine.py b/research/lab/engine.py index 6a8891d..5ae592a 100644 --- a/research/lab/engine.py +++ b/research/lab/engine.py @@ -124,18 +124,46 @@ def select_top(score_row: np.ndarray, top_n: int, held: np.ndarray | None = None return np.array(chosen, int) +def zvol_weights(score_row: np.ndarray, idx: np.ndarray, vol: np.ndarray, + lo: float = 0.5, hi: float = 2.0) -> np.ndarray: + """Grinold (1994) alpha = vol x IC x z, so mean-variance weights with a + diagonal risk model are proportional to z / vol. z is the normal score of + the name's cross-sectional percentile; weights are clipped to + [lo/N, hi/N] and renormalized.""" + from scipy.stats import norm + valid = ~np.isnan(score_row) + pct = np.full(len(score_row), np.nan) + order = np.argsort(score_row[valid], kind="stable") + ranks = np.empty(valid.sum()); ranks[order] = (np.arange(valid.sum()) + 0.5) / valid.sum() + pct[valid] = ranks + z = norm.ppf(pct[idx]) + v = np.where(vol > 0, vol, np.nan) + v = np.nan_to_num(v, nan=np.nanmedian(v) if np.isfinite(v).any() else 1.0) + raw = np.clip(z, 0.01, None) / v + n = len(idx) + w = raw / raw.sum() + for _ in range(5): + w = np.clip(w, lo / n, hi / n) + w = w / w.sum() + return w + + def simulate(panel: Panel, score: pd.DataFrame | None = None, top_n: int = 10, every: int | str = 5, offset: int = 0, warmup: int = 0, weighting: str = "equal", buffer: float = 0.0, exposure: pd.Series | None = None, exec_at: str = "open", target_weights: pd.DataFrame | None = None, min_trade: float = 0.002, - name: str = "") -> Result: + dyn_n: tuple[int, float] | None = None, name: str = "") -> Result: """Run a long-only top-N strategy. score : higher = better, evaluated at each signal date's close. target_weights : alternative to score — explicit weights per signal date (rows need only exist on signal dates; NaN rows skip). weighting : "equal" | "invvol" (60-day vol) | "score" (score-proportional, clipped >0) + | "zvol" (normal score / 60-day vol, see ``zvol_weights``) + dyn_n : (min_n, threshold): hold every member whose score is at least + ``threshold`` (e.g. a mean factor percentile of 0.8), at least + ``min_n`` and at most ``top_n`` names. exposure : fraction of equity to invest, read on the signal date (e.g. a market-trend or vol-target overlay); rest earns cash_ret. exec_at : "open" → trade at next day's open; "close" → next day's close. @@ -159,7 +187,7 @@ def simulate(panel: Panel, score: pd.DataFrame | None = None, top_n: int = 10, TW = (target_weights.reindex(index=dates, columns=panel.close.columns).values if target_weights is not None else None) expo = exposure.reindex(dates).ffill().fillna(1.0).values if exposure is not None else np.ones(n_days) - if weighting == "invvol": + if weighting in ("invvol", "zvol"): rets = panel.close.pct_change(fill_method=None) vol = rets.rolling(60, min_periods=40).std().values sig_mask = rebalance_dates(dates, every, offset, start=warmup) @@ -231,10 +259,15 @@ def simulate(panel: Panel, score: pd.DataFrame | None = None, top_n: int = 10, else: row = np.where(member[d], S[d], np.nan) held = h > 1e-12 - idx = select_top(row, top_n, held, buffer) + n_pick = top_n + if dyn_n is not None: + n_pick = int(np.clip((row >= dyn_n[1]).sum(), dyn_n[0], top_n)) + idx = select_top(row, n_pick, held, buffer) w = np.zeros(n_assets) - if len(idx) >= min(max(1, top_n // 2), 20): - if weighting == "invvol": + if len(idx) >= min(max(1, n_pick // 2), 20): + if weighting == "zvol": + w[idx] = zvol_weights(row, idx, vol[d, idx]) + elif weighting == "invvol": iv = 1.0 / np.where(vol[d, idx] > 0, vol[d, idx], np.nan) iv = np.nan_to_num(iv, nan=np.nanmean(iv) if np.isfinite(iv).any() else 1.0) w[idx] = iv / iv.sum() diff --git a/research/lab/fetch_us_fundamentals.py b/research/lab/fetch_us_fundamentals.py new file mode 100644 index 0000000..c875a46 --- /dev/null +++ b/research/lab/fetch_us_fundamentals.py @@ -0,0 +1,184 @@ +""" +Fetch point-in-time US fundamentals from SEC EDGAR (XBRL companyfacts) for +every S&P 500 ever-member in the research price panel, plus Yahoo split +history (needed to turn split-adjusted prices back into market caps). + + uv run python -m research.lab.fetch_us_fundamentals + +Writes data/research/us_fund_raw.pkl: + {"facts": DataFrame[ticker, tag, start, end, val, filed, form], + "splits": DataFrame (dates x tickers, split ratio on split days)} + +Every fact keeps its FIRST filing date: a value restated in a later filing is +not visible before that later date, so factors built from "filed" are +point-in-time. XBRL coverage starts in 2009-2011 depending on filer size. +""" + +import os +import pickle +import time +from concurrent.futures import ThreadPoolExecutor + +import pandas as pd +import requests +import yfinance as yf + +from research.lab.fetch_us import _proxy_env + +OUT = os.path.join("data", "research", "us_fund_raw.pkl") +UA = {"User-Agent": "quant-research/0.1 gahow@example.com", "Accept-Encoding": "gzip"} +FORMS = {"10-K", "10-Q", "10-K/A", "10-Q/A", "10-KT", "20-F", "40-F"} + +TAGS = [ + # income statement (durations) + "Revenues", "RevenueFromContractWithCustomerExcludingAssessedTax", + "RevenueFromContractWithCustomerIncludingAssessedTax", "SalesRevenueNet", + "SalesRevenueGoodsNet", "SalesRevenueServicesNet", "RevenuesNetOfInterestExpense", + "CostOfRevenue", "CostOfGoodsAndServicesSold", "CostOfGoodsSold", "CostOfServices", + "GrossProfit", "OperatingIncomeLoss", "NetIncomeLoss", "ProfitLoss", + "NetIncomeLossAvailableToCommonStockholdersBasic", + "IncomeLossFromContinuingOperationsBeforeIncomeTaxesExtraordinaryItemsNoncontrollingInterest", + "ResearchAndDevelopmentExpense", "SellingGeneralAndAdministrativeExpense", + "InterestExpense", "IncomeTaxExpenseBenefit", "EarningsPerShareDiluted", + "WeightedAverageNumberOfDilutedSharesOutstanding", + "DepreciationDepletionAndAmortization", "DepreciationAndAmortization", + # cash flow (year-to-date durations in 10-Qs) + "NetCashProvidedByUsedInOperatingActivities", + "NetCashProvidedByUsedInOperatingActivitiesContinuingOperations", + "PaymentsToAcquirePropertyPlantAndEquipment", "PaymentsForRepurchaseOfCommonStock", + "PaymentsOfDividends", "PaymentsOfDividendsCommonStock", + # balance sheet (instants) + "Assets", "AssetsCurrent", "Liabilities", "LiabilitiesCurrent", + "LiabilitiesAndStockholdersEquity", "StockholdersEquity", + "StockholdersEquityIncludingPortionAttributableToNoncontrollingInterest", + "CashAndCashEquivalentsAtCarryingValue", "LongTermDebt", "LongTermDebtNoncurrent", + "LongTermDebtCurrent", "DebtCurrent", "ShortTermBorrowings", "InventoryNet", + "AccountsReceivableNetCurrent", "AccountsPayableCurrent", "PropertyPlantAndEquipmentNet", + "Goodwill", "RetainedEarningsAccumulatedDeficit", "CommonStockSharesOutstanding", +] +DEI_TAGS = ["EntityCommonStockSharesOutstanding"] + + +def cik_map() -> dict[str, int]: + r = requests.get("https://www.sec.gov/files/company_tickers.json", headers=UA, timeout=60) + r.raise_for_status() + return {v["ticker"].upper().replace(".", "-"): int(v["cik_str"]) for v in r.json().values()} + + +def _facts_for(ticker: str, cik: int) -> pd.DataFrame | None: + url = f"https://data.sec.gov/api/xbrl/companyfacts/CIK{cik:010d}.json" + for attempt in range(4): + try: + r = requests.get(url, headers=UA, timeout=120) + if r.status_code == 404: + return None + r.raise_for_status() + js = r.json() + break + except Exception as exc: # noqa: BLE001 — retry transient SEC errors + print(f" {ticker}: {exc}; retry {attempt + 1}", flush=True) + time.sleep(5 * (attempt + 1)) + else: + return None + rows = [] + for ns, tags in (("us-gaap", TAGS), ("dei", DEI_TAGS)): + facts = js.get("facts", {}).get(ns, {}) + for tag in tags: + if tag not in facts: + continue + for unit, xs in facts[tag]["units"].items(): + if unit not in ("USD", "shares", "USD/shares"): + continue + for x in xs: + if x.get("form") not in FORMS: + continue + rows.append((ticker, tag, x.get("start"), x["end"], float(x["val"]), + x["filed"], x["form"])) + if not rows: + return None + df = pd.DataFrame(rows, columns=["ticker", "tag", "start", "end", "val", "filed", "form"]) + # first filing of each (tag, period) wins: later restatements are not + # knowable on the original date + df = df.sort_values("filed").drop_duplicates(["tag", "start", "end"], keep="first") + return df + + +def fetch_splits(tickers: list[str], chunk: int = 60) -> tuple[pd.DataFrame, pd.DataFrame]: + """Split ratios (on split days, 0 elsewhere) and split-adjusted closes + that are NOT dividend-adjusted; market cap = shares reported at c x + close_sa[t] / (product of split ratios after c).""" + splits, closes = [], [] + + def run(batch): + raw = yf.download(batch, start="2004-01-01", actions=True, auto_adjust=False, + progress=False, group_by="column", threads=True) + lv = raw.columns.get_level_values(0) + if "Stock Splits" in lv: + splits.append(raw["Stock Splits"]) + if "Close" in lv: + closes.append(raw["Close"].dropna(axis=1, how="all")) + time.sleep(2) + + def cat(parts): + s = pd.concat(parts, axis=1) + return s.loc[:, ~s.columns.duplicated(keep="last")] + + for i in range(0, len(tickers), chunk): + run(tickers[i:i + chunk]) + # Yahoo fails whole batches now and then; retry the missing names in small batches + for attempt in range(3): + have = set(cat(closes).columns) + missing = [t for t in tickers if t not in have] + if not missing: + break + print(f" retrying {len(missing)} tickers", flush=True) + for i in range(0, len(missing), 15): + run(missing[i:i + 15]) + return cat(splits).fillna(0.0), cat(closes) + + +def main() -> None: + import sys + with open(os.path.join("data", "research", "us_ohlcv.pkl"), "rb") as fh: + tickers = [t for t in pickle.load(fh)["close"].columns] + if "--yahoo-only" in sys.argv: # refresh splits/closes of an existing file + with open(OUT, "rb") as fh: + saved = pickle.load(fh) + _proxy_env() + saved["splits"], saved["close_sa"] = fetch_splits(tickers) + with open(OUT, "wb") as fh: + pickle.dump(saved, fh) + print(f"splits: {int((saved['splits'] > 0).sum().sum())} events; saved {OUT}") + return + ciks = cik_map() + todo = [(t, ciks[t]) for t in tickers if t in ciks] + print(f"{len(todo)}/{len(tickers)} tickers map to a CIK", flush=True) + + frames, t0 = [], time.time() + + def job(tc): + time.sleep(0.25) # stays well under SEC's 10 requests/second + return _facts_for(*tc) + + with ThreadPoolExecutor(max_workers=3) as ex: + for k, df in enumerate(ex.map(job, todo), 1): + if df is not None: + frames.append(df) + if k % 50 == 0: + print(f" {k}/{len(todo)} companies [{time.time() - t0:.0f}s]", flush=True) + facts = pd.concat(frames, ignore_index=True) + for c in ("start", "end", "filed"): + facts[c] = pd.to_datetime(facts[c]) + print(f"{len(facts):,} facts for {facts['ticker'].nunique()} tickers", flush=True) + + _proxy_env() + splits, close_sa = fetch_splits(tickers) + print(f"splits: {int((splits > 0).sum().sum())} events", flush=True) + with open(OUT, "wb") as fh: + pickle.dump({"facts": facts, "splits": splits, "close_sa": close_sa, + "unmapped": sorted(set(tickers) - set(ciks))}, fh) + print(f"Saved {OUT}") + + +if __name__ == "__main__": + main() diff --git a/research/lab/fund_eval.py b/research/lab/fund_eval.py new file mode 100644 index 0000000..7915a03 --- /dev/null +++ b/research/lab/fund_eval.py @@ -0,0 +1,120 @@ +""" +Evaluate the EDGAR fundamental factors (research/lab/fundamentals.py) on +S&P 500 PIT members, 2011-2026: rank IC, long-only top-N backtests, and +blends with the price champion. + + PYTHONPATH=. uv run python -m research.lab.fund_eval +""" + +import argparse +import warnings + +import numpy as np +import pandas as pd + +from research.lab import fundamentals as fu +from research.lab.factor_scan import load, phase_avg +from research.lab.report import forward_returns, ic_series, ic_summary +from research.lab.rotation import factor_ranks + +warnings.filterwarnings("ignore") + +SUBS = [("2011-01-01", "2014-12-31"), ("2015-01-01", "2018-12-31"), + ("2019-01-01", "2022-12-31"), ("2023-01-01", "2026-09-24")] +FIN_SECTORS = {"Financials", "Real Estate"} + + +def rank(df, p, sector_neutral=False): + df = df.where(p.member) + if not sector_neutral: + return df.rank(axis=1, pct=True) + sec = p.extra["sector"] + out = pd.DataFrame(np.nan, index=df.index, columns=df.columns) + for s in sec.dropna().unique(): + cols = [c for c in sec.index[sec == s] if c in df.columns] + out[cols] = df[cols].rank(axis=1, pct=True) + return out + + +def mean_rank(parts): + s = pd.concat(parts, keys=range(len(parts))).groupby(level=1).mean() + return s + + +def blend(*dfs): + stack = np.stack([d.values for d in dfs]) + with np.errstate(invalid="ignore"): + m = np.nanmean(stack, axis=0) + return pd.DataFrame(m, index=dfs[0].index, columns=dfs[0].columns) + + +def evaluate(p, score, **kw): + per = [phase_avg(p, score, a, b, **kw) for a, b in SUBS] + full = phase_avg(p, score, SUBS[0][0], SUBS[-1][1], **kw) + return full, per + + +def line(name, full, per): + return (f"{name:34s} sh {full['sharpe']:.2f} cagr {full['cagr']*100:5.1f}% mdd {full['mdd']*100:5.1f}% " + f"turn {full['turnover']:4.1f}x min {min(x['sharpe'] for x in per):.2f} " + f"[{' '.join(f'{x['sharpe']:.2f}' for x in per)}]") + + +def main() -> None: + ap = argparse.ArgumentParser() + ap.add_argument("--top-n", type=int, default=20) + ap.add_argument("--buffer", type=float, default=1.0) + ap.add_argument("--every", type=int, default=10) + args = ap.parse_args() + kw = dict(top_n=args.top_n, every=args.every, buffer=args.buffer) + p, _ = load("us") + F = fu.load(p) + pr = factor_ranks(p, "us") + fwd20 = forward_returns(p, 20) + a, b = SUBS[0][0], SUBS[-1][1] + names = [k for k in F if k != "mcap"] + R, RS = {}, {} + print("=== single factors (IC20 and top-N), 2011-2026") + for n in names: + R[n] = rank(F[n], p) + RS[n] = rank(F[n], p, sector_neutral=True) + ic = ic_summary(ic_series(F[n].loc[a:b], fwd20.loc[a:b], p.member.loc[a:b], 20), 12) + full, per = evaluate(p, R[n], **kw) + fs, ps = evaluate(p, RS[n], **kw) + print(f"{line(n, full, per)} | IC20 {ic['ic']:+.3f} t {ic['t']:+.1f} | secneu sh {fs['sharpe']:.2f} " + f"cagr {fs['cagr']*100:5.1f}%", flush=True) + R["size_big"] = rank(F["mcap"], p) + full, per = evaluate(p, R["size_big"], **kw) + print(line("size_big (largest caps)", full, per)) + ew = p.close.notna().astype(float).where(p.member) + full, per = evaluate(p, ew, top_n=10000, every=10) + print(line("EW universe", full, per)) + + print("=== composites") + Q = blend(RS["gp_a"], RS["cfo_a"], RS["roe"], RS["accruals"], RS["leverage"]) + V = blend(RS["ep"], RS["cfp"], RS["fcfp"]) + QV = blend(Q, V) + champ = blend(pr["overnight_mom"], pr["rec_126"]) + mom = blend(pr["mom_12_1"], pr["resid_mom"]) + lowrisk = blend(pr["ivol"], pr["low_beta"]) + combos = { + "quality (gp,cfo,roe,accr,lev)": Q, + "value (ep,cfp,fcfp)": V, + "quality+value": QV, + "champ overnight+rec": champ, + "champ + quality": blend(champ, Q), + "champ + quality + value": blend(champ, Q, V), + "champ + qv(1 slot)": blend(champ, QV), + "mom(12_1,resid) + quality": blend(mom, Q), + "mom + quality + value": blend(mom, Q, V), + "mom + quality + lowrisk": blend(mom, Q, lowrisk), + "champ + gp_a": blend(pr["overnight_mom"], pr["rec_126"], RS["gp_a"]), + "champ + shareholder_yield": blend(pr["overnight_mom"], pr["rec_126"], RS["shareholder_yield"]), + } + for name, s in combos.items(): + full, per = evaluate(p, s.where(p.member), **kw) + print(line(name, full, per), flush=True) + + +if __name__ == "__main__": + main() diff --git a/research/lab/fundamentals.py b/research/lab/fundamentals.py new file mode 100644 index 0000000..b842260 --- /dev/null +++ b/research/lab/fundamentals.py @@ -0,0 +1,292 @@ +""" +Point-in-time fundamental factors for the US panel from SEC EDGAR facts +(research/lab/fetch_us_fundamentals.py). + +A value is usable from the trading day AFTER its first filing date, so a +factor on date t only uses statements filed before t. Durations become +quarterly values (direct 3-month facts, or differences of year-to-date facts +that share a fiscal-year start) and trailing-twelve-month sums; instants are +taken as of their period end. + + from research.lab import fundamentals as fu + F = fu.load(panel) # {name: dates x tickers}, cached +""" + +from __future__ import annotations + +import os +import pickle +import time + +import numpy as np +import pandas as pd + +RAW = os.path.join("data", "research", "us_fund_raw.pkl") +CACHE = os.path.join("data", "research", "us_fund_factors.pkl") + +DURATION = { + "revenue": ["Revenues", "RevenueFromContractWithCustomerExcludingAssessedTax", + "RevenueFromContractWithCustomerIncludingAssessedTax", "SalesRevenueNet", + "RevenuesNetOfInterestExpense", "SalesRevenueGoodsNet", "SalesRevenueServicesNet"], + "cogs": ["CostOfRevenue", "CostOfGoodsAndServicesSold", "CostOfGoodsSold", "CostOfServices"], + "gross_profit": ["GrossProfit"], + "op_income": ["OperatingIncomeLoss"], + "net_income": ["NetIncomeLoss", "ProfitLoss", "NetIncomeLossAvailableToCommonStockholdersBasic"], + "cfo": ["NetCashProvidedByUsedInOperatingActivities", + "NetCashProvidedByUsedInOperatingActivitiesContinuingOperations"], + "capex": ["PaymentsToAcquirePropertyPlantAndEquipment"], + "rnd": ["ResearchAndDevelopmentExpense"], + "buyback": ["PaymentsForRepurchaseOfCommonStock"], + "dividends": ["PaymentsOfDividendsCommonStock", "PaymentsOfDividends"], +} +INSTANT = { + "assets": ["Assets"], + "assets_cur": ["AssetsCurrent"], + "liab": ["Liabilities"], + "liab_cur": ["LiabilitiesCurrent"], + "equity": ["StockholdersEquity", + "StockholdersEquityIncludingPortionAttributableToNoncontrollingInterest"], + "ltd": ["LongTermDebtNoncurrent", "LongTermDebt"], + "shares": ["EntityCommonStockSharesOutstanding", "CommonStockSharesOutstanding"], +} + + +# --------------------------------------------------------------------------- +# fact tables → per-ticker event series +# --------------------------------------------------------------------------- + +def _pick(facts: pd.DataFrame, tags: list[str]) -> pd.DataFrame: + """One value per (ticker, start, end): the highest-priority tag present.""" + sub = facts[facts["tag"].isin(tags)].copy() + sub["prio"] = sub["tag"].map({t: i for i, t in enumerate(tags)}) + sub = sub.sort_values(["prio", "filed"]) + return sub.drop_duplicates(["ticker", "start", "end"], keep="first").drop(columns="prio") + + +def _quarters(df: pd.DataFrame) -> pd.DataFrame: + """Quarterly values from duration facts of one ticker: direct ~3-month + facts, else differences of year-to-date facts sharing a start date.""" + df = df.dropna(subset=["start"]) + days = (df["end"] - df["start"]).dt.days + direct = df[(days >= 75) & (days <= 105)][["end", "val", "filed"]] + derived = [] + for _, g in df[days > 105].groupby("start"): + g = g.sort_values("end") + # pair each YTD fact with the one ending ~3 months earlier (same start) + allg = pd.concat([g, df[(df["start"] == g["start"].iloc[0]) & (days.loc[df.index] <= 105)]]) + allg = allg.sort_values("end").drop_duplicates("end") + e = allg["end"].values + v = allg["val"].values + f = allg["filed"].values + for k in range(1, len(allg)): + gap = (e[k] - e[k - 1]).astype("timedelta64[D]").astype(int) + if 75 <= gap <= 105: + derived.append((e[k], v[k] - v[k - 1], max(f[k], f[k - 1]))) + der = pd.DataFrame(derived, columns=["end", "val", "filed"]) + q = pd.concat([direct.assign(src=0), der.assign(src=1)]).sort_values(["end", "src", "filed"]) + q = q.drop_duplicates("end", keep="first").drop(columns="src").sort_values("end") + # the first quarter of a fiscal year never has a same-start predecessor; + # its direct fact covers it + return q.reset_index(drop=True) + + +def _ttm(q: pd.DataFrame) -> pd.DataFrame: + """Sum of four consecutive quarters (ends 75-105 days apart).""" + if len(q) < 4: + return q.iloc[:0] + e = q["end"].values + gaps = np.diff(e).astype("timedelta64[D]").astype(int) + ok_gap = np.r_[False, (gaps >= 75) & (gaps <= 105)] + run = pd.Series(ok_gap).rolling(3).sum().values == 3 # 3 good gaps → 4 quarters + vals = q["val"].rolling(4).sum().values + filed = pd.to_datetime(q["filed"].astype("int64").rolling(4).max().values) + out = pd.DataFrame({"end": e, "val": vals, "filed": filed}) + return out[run].reset_index(drop=True) + + +def _asof_panel(events: dict[str, pd.DataFrame], dates: pd.DatetimeIndex, + tickers: list[str], lag: int = 0) -> pd.DataFrame: + """Daily panel of the value with the latest period end among facts filed + strictly before each date (plus ``lag`` extra periods back: lag=4 gives + the value four quarters earlier than the latest).""" + out = pd.DataFrame(np.nan, index=dates, columns=tickers) + for t, ev in events.items(): + if t not in out.columns or ev.empty: + continue + ev = ev.sort_values("filed") + # as of each filing: the known history sorted by end + rows = [] + known: dict = {} + for end, val, filed in zip(ev["end"].values, ev["val"].values, ev["filed"].values): + known[end] = val + ends = sorted(known) + if len(ends) > lag: + rows.append((filed, known[ends[-1 - lag]])) + if not rows: + continue + s = pd.DataFrame(rows, columns=["filed", "val"]).drop_duplicates("filed", keep="last") + s = s.set_index("filed")["val"] + # usable from the next session after the filing date + pos = dates.searchsorted(s.index, side="right") + ser = pd.Series(np.nan, index=dates) + valid = pos < len(dates) + ser.iloc[pos[valid]] = s.values[valid] + out[t] = ser.ffill().values + return out + + +def _series_by_ticker(df: pd.DataFrame) -> dict[str, pd.DataFrame]: + return {t: g for t, g in df.groupby("ticker")} + + +# --------------------------------------------------------------------------- +# build +# --------------------------------------------------------------------------- + +def build(p) -> dict[str, pd.DataFrame]: + with open(RAW, "rb") as fh: + raw = pickle.load(fh) + facts = raw["facts"] + dates, tickers = p.dates, list(p.close.columns) + t0 = time.time() + Q, TTM, INS = {}, {}, {} + for c, tags in DURATION.items(): + picked = _pick(facts, tags) + q_ev, ttm_ev = {}, {} + for t, g in picked.groupby("ticker"): + q = _quarters(g) + q_ev[t] = q + ttm = _ttm(q) + # fiscal-year facts give TTM at year ends directly + days = (g["end"] - g["start"]).dt.days + fy = g[(days >= 350) & (days <= 380)][["end", "val", "filed"]] + ttm_ev[t] = (pd.concat([fy.assign(src=0), ttm.assign(src=1)]) + .sort_values(["end", "src"]).drop_duplicates("end").drop(columns="src")) + Q[c], TTM[c] = q_ev, ttm_ev + print(f" {c}: {len(q_ev)} tickers [{time.time() - t0:.0f}s]", flush=True) + for c, tags in INSTANT.items(): + if c == "shares": + # cover-page and balance-sheet share counts have different dates; + # mixing them would break "four reports back", so use one tag per ticker + sub = facts[facts["tag"].isin(tags)] + INS[c] = {} + for t, g in sub.groupby("ticker"): + dei = g[g["tag"] == tags[0]] + use = dei if len(dei) >= 8 else g[g["tag"] == tags[1]] + INS[c][t] = use.drop_duplicates("end")[["end", "val", "filed"]] + continue + picked = _pick(facts, tags) + INS[c] = {t: g[["end", "val", "filed"]] for t, g in picked.groupby("ticker")} + + def ttm(c, lag=0): + return _asof_panel(TTM[c], dates, tickers, lag) + + def qv(c, lag=0): + return _asof_panel(Q[c], dates, tickers, lag) + + def ins(c, lag=0): + return _asof_panel(INS[c], dates, tickers, lag) + + F: dict[str, pd.DataFrame] = {} + rev, ni, cfo = ttm("revenue"), ttm("net_income"), ttm("cfo") + gp = ttm("gross_profit") + cogs = ttm("cogs") + gp = gp.combine_first(rev - cogs) + opi = ttm("op_income") + capex = ttm("capex") + assets, equity, liab = ins("assets"), ins("equity"), ins("liab") + assets_4 = ins("assets", 4) + print(f" panels [{time.time() - t0:.0f}s]", flush=True) + pos = lambda x: x.where(x > 0) # noqa: E731 + + F["gp_a"] = gp / pos(assets) # Novy-Marx 2013 + F["op_a"] = opi / pos(assets) + F["cfo_a"] = cfo / pos(assets) # cash profitability + F["roe"] = ni / pos(equity) + F["roa"] = ni / pos(assets) + F["accruals"] = -(ni - cfo) / pos(assets) # Sloan 1996 (low is good) + F["asset_growth"] = -(assets / pos(assets_4) - 1) # Cooper-Gulen-Schill 2008 + F["leverage"] = -(liab / pos(assets)) + F["gross_margin"] = gp / pos(rev) + rev_4 = ttm("revenue", 4) + F["rev_growth"] = rev / pos(rev_4) - 1 + gm_4 = ttm("gross_profit", 4).combine_first(rev_4 - ttm("cogs", 4)) / pos(rev_4) + F["gm_change"] = F["gross_margin"] - gm_4 + ni_4 = ttm("net_income", 4) + F["roa_change"] = F["roa"] - ni_4 / pos(assets_4) + + # SUE: latest quarterly change vs same quarter last year, scaled by the + # std of the last 8 such changes (seasonal random walk, Bernard-Thomas) + for c, name in (("net_income", "sue"), ("revenue", "sue_rev")): + chg = {} + for t, q in Q[c].items(): + q = q.sort_values("end").reset_index(drop=True) + if len(q) < 8: + continue + e = q["end"].values + v = q["val"].values + d = np.full(len(q), np.nan) + for k in range(4, len(q)): + gap = (e[k] - e[k - 4]).astype("timedelta64[D]").astype(int) + if 350 <= gap <= 380: + d[k] = v[k] - v[k - 4] + ds = pd.Series(d) + sd = ds.rolling(8, min_periods=4).std() + sue = (ds / sd.replace(0, np.nan)).values + chg[t] = pd.DataFrame({"end": e, "val": sue, "filed": q["filed"].values}).dropna() + F[name] = _asof_panel(chg, dates, tickers).clip(-10, 10) + + # market cap. Yahoo's split-adjusted close is raw / F(t), where F(t) is the + # product of split ratios after t, and shares reported at c grow by the + # splits in (c, t], so cap_t = shares_c * F(c) * close_sa_t. + splits = raw["splits"].reindex(columns=tickers).fillna(0.0) + close_sa = raw["close_sa"].reindex(index=dates, columns=tickers).ffill() + ratio = splits.replace(0.0, 1.0) + # F(c) = product of split ratios strictly after c + after = ratio.iloc[::-1].cumprod().iloc[::-1].shift(-1).fillna(1.0) + after = after.reindex(dates, method="ffill").fillna(1.0) + sh_ev = {} + for t, g in INS["shares"].items(): + if t not in after.columns: + continue + ff = after[t].reindex(g["end"].values, method="ffill").fillna(1.0).values + sh_ev[t] = g.assign(val=g["val"].values * ff) # shares on today's split basis + shares_now = _asof_panel(sh_ev, dates, tickers) + mcap = shares_now * close_sa + mcap = mcap.where(mcap > 1e8) + F["mcap"] = mcap + F["ep"] = ni / mcap + F["cfp"] = cfo / mcap + F["sp"] = rev / mcap + F["bm"] = equity / mcap + F["fcfp"] = (cfo - capex.fillna(0)) / mcap + payout = ttm("buyback").fillna(0) + ttm("dividends").fillna(0) + F["shareholder_yield"] = payout / mcap + sh_4 = _asof_panel(sh_ev, dates, tickers, lag=4) + F["net_issuance"] = -np.log(shares_now / sh_4).clip(-1, 1) + + # Piotroski F-score (9 binary signals) + ltd, ac, lc = ins("ltd"), ins("assets_cur"), ins("liab_cur") + ltd_4, ac_4, lc_4 = ins("ltd", 4), ins("assets_cur", 4), ins("liab_cur", 4) + sig = [ + F["roa"] > 0, F["cfo_a"] > 0, F["roa_change"] > 0, cfo > ni, + (ltd / pos(assets)) < (ltd_4 / pos(assets_4)), + (ac / pos(lc)) > (ac_4 / pos(lc_4)), + F["net_issuance"] >= -0.005, + F["gm_change"] > 0, + (rev / pos(assets)) > (rev_4 / pos(assets_4)), + ] + avail = F["roa"].notna() & F["cfo_a"].notna() + F["fscore"] = sum(s.astype(float) for s in sig).where(avail) + print(f" factors built [{time.time() - t0:.0f}s]", flush=True) + return {k: v.astype("float32") for k, v in F.items()} + + +def load(p, rebuild: bool = False) -> dict[str, pd.DataFrame]: + if os.path.exists(CACHE) and not rebuild: + with open(CACHE, "rb") as fh: + return pickle.load(fh) + F = build(p) + with open(CACHE, "wb") as fh: + pickle.dump(F, fh, protocol=4) + return F diff --git a/research/lab/overlays.py b/research/lab/overlays.py index 669b316..24016bc 100644 --- a/research/lab/overlays.py +++ b/research/lab/overlays.py @@ -53,3 +53,172 @@ def vol_target(p, score, target: float, lookback: int = 60, cap: float = 1.0, if exposure is not None: scale = scale * exposure.reindex(scale.index).ffill().fillna(1.0) return scale + + +# --------------------------------------------------------------------------- +# Literature overlays on an index level (all read at the close of the signal +# date; engine.simulate applies them at the next rebalance) +# --------------------------------------------------------------------------- + +def faber_sma(level: pd.Series, months: int = 10) -> pd.Series: + """Faber (2007): invested while the month-end close is above its + ``months``-month simple average; the state changes only at month ends.""" + m = level.resample("ME").last() + on = (m > m.rolling(months, min_periods=months).mean()).astype(float) + on = on.where(m.rolling(months).count() >= months, 1.0) + return on.reindex(level.index, method="ffill").fillna(1.0) + + +def derisk_vol(level: pd.Series, halflife: int = 20, min_hist: int = 252) -> pd.Series: + """De-risk-only volatility target (Harvey et al. 2018 estimator): exposure + = min(1, median vol to date / EWMA vol), so it only ever cuts.""" + r = level.pct_change(fill_method=None).fillna(0.0) + v = np.sqrt((r ** 2).ewm(halflife=halflife, min_periods=halflife).mean()) * np.sqrt(252) + med = v.expanding(min_periods=min_hist).median() + return (med / v).clip(upper=1.0).fillna(1.0) + + +def momentum_state(level: pd.Series, months: int = 24, cut: float = 0.5) -> pd.Series: + """Daniel & Moskowitz (2016) / Cooper et al. (2004): after a negative + ``months``-month market return momentum is crash-prone; hold ``cut``.""" + m = level.resample("ME").last() + down = (m / m.shift(months) - 1) < 0 + return down.map({True: cut, False: 1.0}).reindex(level.index, method="ffill").fillna(1.0) + + +# --- statistical jump model (Shu, Yu & Mulvey 2024, arXiv:2402.05272) ------- + +def jm_features(r: pd.Series) -> pd.DataFrame: + """EWM downside deviation (halflife 10) and EWM Sortino ratios + (halflives 20 and 60) of daily excess returns.""" + feats = {} + dd10 = np.sqrt((np.minimum(r, 0) ** 2).ewm(halflife=10).mean()) + feats["dd10"] = np.log(dd10 + 1e-8) + for hl in (20, 60): + mu = r.ewm(halflife=hl).mean() + dd = np.sqrt((np.minimum(r, 0) ** 2).ewm(halflife=hl).mean()) + feats[f"sortino{hl}"] = (mu / (dd + 1e-8)).clip(-10, 10) + return pd.DataFrame(feats) + + +try: # the DP loops are ~50x faster compiled + from numba import njit +except ImportError: # pragma: no cover - pure-Python fallback + def njit(f=None, **kw): + return f if f is not None else (lambda g: g) + + +@njit(cache=True) +def _jm_viterbi(cost: np.ndarray, lam: float) -> np.ndarray: + T = cost.shape[0] + V0, V1 = cost[0, 0], cost[0, 1] + back = np.zeros((T, 2), np.int64) + for t in range(1, T): + a0, b0 = V0, V1 + lam # into state 0: stay / switch + a1, b1 = V1, V0 + lam + back[t, 0] = 0 if a0 <= b0 else 1 + back[t, 1] = 1 if a1 <= b1 else 0 + V0, V1 = cost[t, 0] + min(a0, b0), cost[t, 1] + min(a1, b1) + s = np.empty(T, np.int64) + s[T - 1] = 0 if V0 <= V1 else 1 + for t in range(T - 1, 0, -1): + s[t - 1] = back[t, s[t]] + return s + + +@njit(cache=True) +def _jm_forward(cost: np.ndarray, lam: float) -> np.ndarray: + """State at each t from a forward DP over data through t (no look-ahead).""" + T = cost.shape[0] + V0, V1 = cost[0, 0], cost[0, 1] + out = np.empty(T, np.int64) + out[0] = 0 if V0 <= V1 else 1 + for t in range(1, T): + V0, V1 = cost[t, 0] + min(V0, V1 + lam), cost[t, 1] + min(V1, V0 + lam) + out[t] = 0 if V0 <= V1 else 1 + return out + + +def _jm_cost(X: np.ndarray, mu: np.ndarray) -> np.ndarray: + return ((X[:, None, :] - mu[None]) ** 2).sum(axis=2) + + +def _jm_fit(X: np.ndarray, lam: float, n_init: int = 6, iters: int = 30, seed: int = 0): + """Two-state jump model: min sum ||x_t - mu_{s_t}||^2 + lam * [s_t != s_{t-1}] + by coordinate descent (dynamic programming for states, means for centers).""" + rng = np.random.default_rng(seed) + best = (np.inf, None) + for _ in range(n_init): + mu = X[rng.choice(len(X), 2, replace=False)] + for _ in range(iters): + s = _jm_viterbi(_jm_cost(X, mu), lam) + new = np.array([X[s == k].mean(axis=0) if (s == k).any() else mu[k] for k in (0, 1)]) + if np.allclose(new, mu): + break + mu = new + s = _jm_viterbi(_jm_cost(X, mu), lam) + obj = _jm_cost(X, mu)[np.arange(len(X)), s].sum() + lam * np.abs(np.diff(s)).sum() + if obj < best[0]: + best = (obj, mu.copy()) + return best[1] + + +def _jm_online(X: np.ndarray, mu: np.ndarray, lam: float) -> np.ndarray: + return _jm_forward(_jm_cost(X, mu), lam) + + +def jump_model(level: pd.Series, cash_ret: pd.Series | None = None, + lams: tuple[float, ...] = (0.0, 5.0, 10.0, 20.0, 30.0, 50.0, 80.0, 120.0), + train_days: int = 3000, min_train: int = 1500, refit_months: int = 6, + select_years: int = 8, min_select_years: int = 4, cost: float = 0.001) -> pd.Series: + """0/1 exposure from the jump model. Centers are refit every + ``refit_months`` on the trailing ``train_days`` (at least ``min_train``); + features are standardized with the training window's moments; the bull + state is the one with the higher mean return in training. Each month the + jump penalty is re-chosen as the one whose online 0/1 strategy (1-day + delay, ``cost`` per switch) had the best Sharpe over the trailing + ``select_years`` (at least ``min_select_years``).""" + r = level.pct_change(fill_method=None).fillna(0.0) + cr = cash_ret.reindex(r.index).fillna(0.0) if cash_ret is not None else pd.Series(0.0, index=r.index) + ex = r - cr + F = jm_features(ex) + dates = r.index + refits = [d for d in pd.date_range(dates[0], dates[-1], freq=f"{refit_months}MS") + if dates.searchsorted(d) >= min_train] + states = {} + for lam in lams: + st = pd.Series(np.nan, index=dates) + for i, d in enumerate(refits): + e = dates.searchsorted(d) + s0 = max(0, e - train_days) + Xtr = F.iloc[s0:e] + m, sd = Xtr.mean(), Xtr.std().replace(0, 1) + mu = _jm_fit(((Xtr - m) / sd).values, lam) + nxt = dates.searchsorted(refits[i + 1]) if i + 1 < len(refits) else len(dates) + # run the forward DP from the window start so the state is warm + Xall = ((F.iloc[s0:nxt] - m) / sd).values + path = _jm_online(Xall, mu, lam) + tr_states = path[:e - s0] + bull = int(np.argmax([ex.iloc[s0:e][tr_states == k].mean() if (tr_states == k).any() else -1 + for k in (0, 1)])) + st.iloc[e:nxt] = (path[e - s0:] == bull).astype(float) + states[lam] = st + S = pd.DataFrame(states) + # returns of each penalty's 0/1 strategy, held from the next day + pos = S.shift(1) + strat = pos.mul(r, axis=0) + (1 - pos).mul(cr, axis=0) - pos.diff().abs() * cost + month_ends = strat.resample("ME").last().index + choice = pd.Series(np.nan, index=dates) + for me in month_ends: + hist = strat.loc[:me].dropna() + hist = hist[hist.index > me - pd.DateOffset(years=select_years)] + if len(hist) < 252 * min_select_years: + continue + sh = (hist.mean() / hist.std()).dropna() + if sh.empty: + continue + lam = sh.idxmax() + nxt = dates.searchsorted(me, side="right") + end = dates.searchsorted(me + pd.offsets.MonthEnd(1), side="right") + choice.iloc[nxt:end] = S[lam].iloc[nxt:end].values + return choice.fillna(1.0) diff --git a/research/lab/persistence.py b/research/lab/persistence.py new file mode 100644 index 0000000..55483a3 --- /dev/null +++ b/research/lab/persistence.py @@ -0,0 +1,98 @@ +""" +Does a factor combo that did well recently keep doing well? + +For every pair of price factors (equal-weight rank composite) compute a cheap +daily proxy: the equal-weight return of its top-N members on the previous +close's scores, minus the member average. Then, each calendar year, rank the +pairs by their trailing 1/3/5-year proxy Sharpe and look at how the chosen +pairs do the NEXT year, against the average pair and a fixed pick. + + PYTHONPATH=. uv run python -m research.lab.persistence +""" + +from __future__ import annotations + +import itertools +import os +import pickle +import time +import warnings + +import numpy as np +import pandas as pd +from scipy import stats as sps + +from research.lab.factor_scan import RESULTS, load +from research.lab.rotation import factor_ranks + +warnings.filterwarnings("ignore") + + +def top_proxy(p, score: pd.DataFrame, top_n: int = 20) -> pd.Series: + """Daily excess return of the top-N (previous close scores) over members.""" + S = score.shift(1).values + r = p.close.pct_change(fill_method=None).values + mem = p.member.shift(1, fill_value=False).values + S = np.where(mem & ~np.isnan(r), S, np.nan) + out = np.full(len(S), np.nan) + base = np.nanmean(np.where(mem, r, np.nan), axis=1) + for t in range(len(S)): + row = S[t] + ok = np.where(~np.isnan(row))[0] + if len(ok) < top_n * 3: + continue + top = ok[np.argpartition(-row[ok], top_n)[:top_n]] + out[t] = np.nanmean(r[t, top]) - base[t] + return pd.Series(out, index=p.dates) + + +def main() -> None: + p, _ = load("us") + ranks = factor_ranks(p, "us") + names = sorted(ranks) + path = os.path.join(RESULTS, "pair_proxies_us.pkl") + if os.path.exists(path): + with open(path, "rb") as fh: + P = pickle.load(fh) + else: + t = time.time() + P = {} + for a, b in itertools.combinations(names, 2): + s = (ranks[a] + ranks[b]) / 2 + P[f"{a}+{b}"] = top_proxy(p, s) + for a in names: + P[a] = top_proxy(p, ranks[a]) + P = pd.DataFrame(P) + print(f"{P.shape[1]} proxies [{time.time() - t:.0f}s]", flush=True) + with open(path, "wb") as fh: + pickle.dump(P, fh) + P = P.loc["2006":] + yearly = P.groupby(P.index.year).apply(lambda x: x.mean() / x.std() * np.sqrt(252)) # excess Sharpe (IR) + yret = P.groupby(P.index.year).sum() # ~excess return + years = [y for y in yearly.index if y >= 2008 and y <= 2025] + print("\nRank correlation of combo IR between consecutive years (all pairs):") + rho = [sps.spearmanr(yearly.loc[y - 1], yearly.loc[y], nan_policy="omit").statistic for y in years] + print(" " + " ".join(f"{y}:{r:+.2f}" for y, r in zip(years, rho)), f"| mean {np.nanmean(rho):+.3f}") + for look in (1, 3, 5): + rows = [] + for y in range(2006 + look, 2027): + hist = P.loc[str(y - look):str(y - 1)] + ir = hist.mean() / hist.std() + nxt = yret.loc[y] if y in yret.index else None + if nxt is None: + continue + best = ir.nlargest(10).index + worst = ir.nsmallest(10).index + rows.append({"year": y, "top10": nxt[best].mean(), "bottom10": nxt[worst].mean(), + "all": nxt.mean(), "best1": nxt[ir.idxmax()], "best1_name": ir.idxmax()}) + df = pd.DataFrame(rows).set_index("year") + print(f"\nlookback {look}y: next-year excess return of the pairs ranked by trailing IR") + print(f" mean top10 {df['top10'].mean()*100:+.2f}% bottom10 {df['bottom10'].mean()*100:+.2f}% " + f"all {df['all'].mean()*100:+.2f}% best1 {df['best1'].mean()*100:+.2f}% " + f"top10 beats all in {(df['top10'] > df['all']).mean()*100:.0f}% of years") + if look == 1: + print(df.assign(**{c: (df[c] * 100).round(1) for c in ["top10", "bottom10", "all", "best1"]}).to_string()) + + +if __name__ == "__main__": + main() diff --git a/research/lab/rotation.py b/research/lab/rotation.py new file mode 100644 index 0000000..918f376 --- /dev/null +++ b/research/lab/rotation.py @@ -0,0 +1,258 @@ +""" +Walk-forward factor weighting: does choosing factors from their recent record +beat holding a fixed composite? + +Each scheme sets factor weights at every month end from data available then +(trailing factor returns / ICs); the stock score is the weighted average of +the factors' cross-sectional percentile ranks. The resulting scores run +through the same engine (PIT members, next-open fills, phase averaging). + + PYTHONPATH=. uv run python -m research.lab.rotation --market us +""" + +from __future__ import annotations + +import argparse +import os +import pickle +import time +import warnings + +import numpy as np +import pandas as pd + +from research.lab.factor_scan import RESULTS, load, phase_avg +from research.lab.factors import FACTORS +from research.lab.report import forward_returns + +warnings.filterwarnings("ignore") + + +# --------------------------------------------------------------------------- +# inputs +# --------------------------------------------------------------------------- + +def factor_ranks(p, market: str, names: list[str] | None = None) -> dict[str, pd.DataFrame]: + """Percentile ranks among PIT members, cached per market (float32).""" + path = os.path.join("data", "research", f"ranks_{market}.pkl") + cache = {} + if os.path.exists(path): + with open(path, "rb") as fh: + cache = pickle.load(fh) + mk = "us" if market == "us" else "cn" + names = names or [n for n, f in FACTORS.items() if mk in f.markets] + dirty = False + for n in names: + if n in cache: + continue + t = time.time() + try: + f = FACTORS[n](p).reindex(index=p.dates, columns=p.close.columns) + except Exception as exc: # noqa: BLE001 + print(f"skip {n}: {exc}", flush=True) + continue + cache[n] = f.where(p.member).rank(axis=1, pct=True).astype("float32") + dirty = True + print(f" rank {n} [{time.time() - t:.0f}s]", flush=True) + if dirty: + with open(path, "wb") as fh: + pickle.dump(cache, fh, protocol=4) + return {n: cache[n] for n in names if n in cache} + + +def factor_returns(p, ranks: dict[str, pd.DataFrame], q: float = 0.2) -> pd.DataFrame: + """Daily return of each factor's top quintile minus the member average, + formed on the previous close's ranks (a timing input, not a tradable P&L).""" + r = p.close.pct_change(fill_method=None) + mem = p.member.shift(1, fill_value=False) + base = r.where(mem).mean(axis=1) + out = {} + for n, rk in ranks.items(): + top = rk.shift(1) >= 1 - q + out[n] = r.where(top & mem).mean(axis=1) - base + return pd.DataFrame(out).fillna(0.0) + + +def monthly_ic(p, ranks: dict[str, pd.DataFrame], horizon: int = 21) -> pd.DataFrame: + """Rank IC of each factor at month ends vs the next ``horizon`` days' open-to-open + return. The IC of month m is only known ``horizon`` days later.""" + fwd = forward_returns(p, horizon).where(p.member).rank(axis=1, pct=True) + me = _month_ends(p.dates) + out = {} + for n, rk in ranks.items(): + a = rk.loc[me] + b = fwd.loc[me] + am = a.sub(a.mean(axis=1), axis=0) + bm = b.sub(b.mean(axis=1), axis=0) + ok = a.notna() & b.notna() + cov = (am * bm).where(ok).sum(axis=1) + out[n] = cov / np.sqrt((am ** 2).where(ok).sum(axis=1) * (bm ** 2).where(ok).sum(axis=1)) + return pd.DataFrame(out) + + +def _month_ends(dates: pd.DatetimeIndex) -> pd.DatetimeIndex: + s = pd.Series(dates, index=dates) + return pd.DatetimeIndex(s.groupby(dates.to_period("M")).max().values) + + +# --------------------------------------------------------------------------- +# weighting schemes: each returns month-end x factor weights (>= 0) +# --------------------------------------------------------------------------- + +def _monthly(fr: pd.DataFrame) -> pd.DataFrame: + return (1 + fr).resample("ME").prod() - 1 + + +def w_equal(fr: pd.DataFrame, names: list[str]) -> pd.DataFrame: + return pd.DataFrame(1.0, index=_monthly(fr).index, columns=names) + + +def w_tsmom(fr: pd.DataFrame, names: list[str], lookback: int = 12) -> pd.DataFrame: + """Time-series factor momentum (Gupta & Kelly 2019; Ehsani & Linnainmaa 2022): + hold a factor when its trailing ``lookback``-month return is positive.""" + m = _monthly(fr[names]) + trail = (1 + m).rolling(lookback, min_periods=lookback).apply(np.prod, raw=True) - 1 + return (trail > 0).astype(float).where(trail.notna()) + + +def w_csmom(fr: pd.DataFrame, names: list[str], lookback: int = 12, k: int = 5) -> pd.DataFrame: + """Cross-sectional factor momentum: the ``k`` factors with the best trailing return.""" + m = _monthly(fr[names]) + trail = (1 + m).rolling(lookback, min_periods=lookback).apply(np.prod, raw=True) - 1 + rank = trail.rank(axis=1, ascending=False) + return (rank <= k).astype(float).where(trail.notna()) + + +def w_ic(ic: pd.DataFrame, names: list[str], lookback: int = 36) -> pd.DataFrame: + """Weight by the trailing mean IC, clipped at zero. The IC of month end m + needs 22 more sessions of prices, so it is used from two month ends later.""" + known = ic[names].shift(2) + mean = known.rolling(lookback, min_periods=lookback // 2).mean() + return mean.clip(lower=0) + + +def w_icir(ic: pd.DataFrame, names: list[str], lookback: int = 36) -> pd.DataFrame: + known = ic[names].shift(2) + roll = known.rolling(lookback, min_periods=lookback // 2) + return (roll.mean() / roll.std()).clip(lower=0) + + +def w_state(fr: pd.DataFrame, names: list[str], state: pd.Series, min_obs: int = 24) -> pd.DataFrame: + """Regime-conditional factor choice: in each month, weight factors whose + average return in past months of the SAME market state was positive. + ``state`` is a month-end label known at that month end.""" + m = _monthly(fr[names]) + st = state.reindex(m.index) + out = pd.DataFrame(np.nan, index=m.index, columns=names) + for i in range(len(m.index)): + # weights set at month end i are used during month i+1; the history + # pairs each month's return with the state at the end of the month before + hist = m.iloc[:i + 1] + prev_state = st.iloc[:i + 1].shift(1) + same = hist[prev_state == st.iloc[i]] + if len(same) >= min_obs: + out.iloc[i] = (same.mean() > 0).astype(float).values + return out + + +def market_state(p) -> pd.Series: + """Month-end state: EW member index above/below its 10-month average x + trailing 1-month vol above/below its 5-year median.""" + r = p.close.pct_change(fill_method=None).where(p.member.shift(1, fill_value=False)).mean(axis=1).fillna(0) + lvl = (1 + r).cumprod() + mlvl = lvl.resample("ME").last() + trend = mlvl > mlvl.rolling(10, min_periods=10).mean() + vol = r.rolling(21).std().resample("ME").last() + hi = vol > vol.rolling(60, min_periods=24).median() + lab = trend.map({True: "up", False: "down"}) + "_" + hi.map({True: "hv", False: "lv"}) + return lab.where(mlvl.rolling(10).count() >= 10) + + +# --------------------------------------------------------------------------- +# blend +# --------------------------------------------------------------------------- + +def blend(p, ranks: dict[str, pd.DataFrame], W: pd.DataFrame) -> pd.DataFrame: + """Daily score = weighted mean of available factor ranks, using the most + recent month-end weights (applied from the next trading day).""" + names = [c for c in W.columns if c in ranks] + Wd = W[names].reindex(p.dates, method="ffill").shift(1) + num = np.zeros(p.close.shape, dtype="float64") + den = np.zeros(p.close.shape, dtype="float64") + for n in names: + R = ranks[n].values + w = np.nan_to_num(Wd[n].values, nan=0.0)[:, None] + ok = ~np.isnan(R) + num += np.where(ok, R, 0.0) * w + den += ok * w + with np.errstate(invalid="ignore", divide="ignore"): + s = num / den + s[den < 1e-9] = np.nan + return pd.DataFrame(s, index=p.dates, columns=p.close.columns).where(p.member) + + +# --------------------------------------------------------------------------- + +EVAL = { + "us": [("2010-01-01", "2013-06-30"), ("2013-07-01", "2016-09-26"), + ("2016-09-27", "2021-09-26"), ("2021-09-27", "2026-09-24")], + "cn": [("2012-01-01", "2014-06-30"), ("2014-07-01", "2016-09-26"), + ("2016-09-27", "2021-09-26"), ("2021-09-27", "2026-09-24")], +} + + +def evaluate(p, score, subs, **kw) -> dict: + per = [phase_avg(p, score, a, b, **kw) for a, b in subs] + full = phase_avg(p, score, subs[0][0], subs[-1][1], **kw) + return {"sharpe": full["sharpe"], "cagr": full["cagr"], "mdd": full["mdd"], + "turn": full["turnover"], "min_block": min(x["sharpe"] for x in per), + "blocks": " ".join(f"{x['sharpe']:.2f}" for x in per)} + + +def main() -> None: + ap = argparse.ArgumentParser() + ap.add_argument("--market", default="us") + ap.add_argument("--top-n", type=int, default=20) + ap.add_argument("--every", type=int, default=10) + ap.add_argument("--buffer", type=float, default=1.0) + args = ap.parse_args() + mk = "us" if args.market == "us" else "cn" + p, _ = load(args.market) + t = time.time() + ranks = factor_ranks(p, args.market) + names = list(ranks) + print(f"{len(names)} factors ranked [{time.time() - t:.0f}s]", flush=True) + fr = factor_returns(p, ranks) + ic = monthly_ic(p, ranks) + with open(os.path.join(RESULTS, f"factor_returns_{args.market}.pkl"), "wb") as fh: + pickle.dump({"fr": fr, "ic": ic}, fh) + state = market_state(p) + schemes = { + "equal_all": w_equal(fr, names), + "tsmom_12": w_tsmom(fr, names, 12), + "tsmom_6": w_tsmom(fr, names, 6), + "tsmom_1": w_tsmom(fr, names, 1), + "csmom_12_top5": w_csmom(fr, names, 12, 5), + "csmom_12_top10": w_csmom(fr, names, 12, 10), + "csmom_1_top5": w_csmom(fr, names, 1, 5), + "csmom_36_top5": w_csmom(fr, names, 36, 5), + "ic_36": w_ic(ic, names, 36), + "ic_60": w_ic(ic, names, 60), + "icir_36": w_icir(ic, names, 36), + "state_cond": w_state(fr, names, state), + } + subs = EVAL[mk] + kw = dict(top_n=args.top_n, every=args.every, buffer=args.buffer) + rows = [] + for name, W in schemes.items(): + s = blend(p, ranks, W) + st = evaluate(p, s, subs, **kw) + rows.append({"scheme": name, **st}) + nact = (W.loc[subs[0][0]:] > 0).sum(axis=1).mean() + print(f"{name:16s} sh {st['sharpe']:.2f} cagr {st['cagr']*100:5.1f}% mdd {st['mdd']*100:5.1f}% " + f"turn {st['turn']:4.1f}x min {st['min_block']:.2f} [{st['blocks']}] active {nact:.1f}", flush=True) + pd.DataFrame(rows).to_csv(os.path.join(RESULTS, f"rotation_{args.market}.csv"), index=False) + + +if __name__ == "__main__": + main() diff --git a/research/lab/trend_rider_long.py b/research/lab/trend_rider_long.py new file mode 100644 index 0000000..409289a --- /dev/null +++ b/research/lab/trend_rider_long.py @@ -0,0 +1,75 @@ +""" +TrendRider equity curves from 2010 (the live ETF panel starts in 2016), for +the US portfolio research (research/lab/us_final.py). + + PYTHONPATH=. uv run python -m research.lab.trend_rider_long + +Downloads the TrendRider ETF basket from 2008 ticker by ticker (batched +Yahoo requests fail intermittently) and runs each variant through the live +book simulation ($2 per order on $10k, 5 bp, 10% resize band). Writes +data/research/tr_long.pkl: {"equity": {name: Series}, "close": DataFrame}. +A ticker that still fails is left out; the strategy then picks among the +rest of its basket (in the 2026-09-29 run DBC was missing, so the risk-off +basket was GLD alone). +""" + +import os +import pickle +import time +import warnings + +import pandas as pd +import yfinance as yf + +from main import simulate_live_book, strategy_positions +from research.lab.fetch_us import _proxy_env +from strategies.permanent import ETF_UNIVERSE +from trader import RESIZE_BAND, STRATEGY_REGISTRY + +warnings.filterwarnings("ignore") +OUT = os.path.join("data", "research", "tr_long.pkl") +VARIANTS = ["trend_rider_v7_vt24", "trend_rider_v7_vt36", "trend_rider_v5_vt30"] +BLOCKS = [("2010-03-01", "2013-02-28"), ("2013-03-01", "2016-09-26"), + ("2016-09-27", "2021-09-26"), ("2021-09-27", "2026-09-25"), ("2010-03-01", "2026-09-25")] + + +def fetch(tickers: list[str], start: str = "2008-01-01") -> tuple[pd.DataFrame, pd.DataFrame]: + close, opens = {}, {} + for t in tickers: + for attempt in range(5): + try: + h = yf.Ticker(t).history(start=start, auto_adjust=True) + if len(h): + close[t], opens[t] = h["Close"], h["Open"] + break + except Exception as exc: # noqa: BLE001 — retry transient Yahoo errors + print(f" {t}: {exc}") + time.sleep(3 * (attempt + 1)) + c, o = pd.DataFrame(close), pd.DataFrame(opens) + c.index = c.index.tz_localize(None).normalize() + o.index = c.index + return c, o + + +def main() -> None: + _proxy_env() + tickers = sorted(set(ETF_UNIVERSE) | {"SPY", "SHY"}) + close, opens = fetch(tickers) + print("missing:", sorted(set(tickers) - set(close.columns))) + out = {} + for name in VARIANTS: + held = strategy_positions(STRATEGY_REGISTRY[name](), close, open_data=opens) + eq, _ = simulate_live_book(held, close, 10_000, 0.0005, 2.0, 0, 0, 50.0, RESIZE_BAND) + out[name] = eq + r = eq.pct_change().dropna() + for a, b in BLOCKS: + x = r.loc[a:b] + e = (1 + x).cumprod() + print(f"{name} {a[:4]}-{b[:4]} cagr {(e.iloc[-1] ** (252 / len(x)) - 1) * 100:5.1f}% " + f"sh {x.mean() / x.std() * 252 ** 0.5:.2f} mdd {((e / e.cummax()) - 1).min() * 100:5.1f}%") + with open(OUT, "wb") as fh: + pickle.dump({"equity": out, "close": close}, fh) + + +if __name__ == "__main__": + main() diff --git a/research/lab/us_final.py b/research/lab/us_final.py new file mode 100644 index 0000000..4b70dd4 --- /dev/null +++ b/research/lab/us_final.py @@ -0,0 +1,155 @@ +""" +Final check of the recommended US portfolio: three sleeves with different +return sources, equal capital, rebalanced monthly between sleeves. + + momentum : overnight-return momentum + distance from 126d low, top 10, 10-day + rebalance, hold while rank <= 30 (the live overnight_recovery_top10) + defensive : sector-neutral quality (GP/A, CFO/A, ROE, accruals, leverage) + + low risk (idiosyncratic vol, beta), top 20, monthly, buffer 2 + (price-only alternative: low risk alone) + trend : TrendRider V7 VT24 on leveraged ETFs (live trend_rider_v7_vt24) + + PYTHONPATH=. uv run python -m research.lab.trend_rider_long # once: data/research/tr_long.pkl + PYTHONPATH=. uv run --with numba python -m research.lab.us_final + +The live version is ``us_three_sleeve`` (strategies/sleeve_portfolio.py). +""" + +from __future__ import annotations + +import json +import os +import pickle +import warnings + +import numpy as np +import pandas as pd + +from research.lab import fundamentals as fu +from research.lab.engine import Costs +from research.lab.factor_scan import RESULTS, load +from research.lab.report import deflated_sharpe, stats +from research.lab.rotation import factor_ranks +from research.lab.us_sweep import _blend, run, sector_rank + +warnings.filterwarnings("ignore") + +SPAN = ("2011-01-03", "2026-09-24") +LONG = ("2006-09-27", "2026-09-24") +BLOCKS = [("2006-09-27", "2010-12-31"), ("2011-01-01", "2014-12-31"), ("2015-01-01", "2018-12-31"), + ("2019-01-01", "2022-12-31"), ("2023-01-01", "2026-09-24")] + + +def sleeve_scores(p): + pr = factor_ranks(p, "us") + F = fu.load(p) + Q = _blend(*(sector_rank(F[k], p) for k in ("gp_a", "cfo_a", "roe", "accruals", "leverage"))) + low = _blend(pr["ivol"], pr["low_beta"]) + return { + "momentum": (_blend(pr["overnight_mom"], pr["rec_126"]), dict(top_n=10, buffer=2.0, every=10)), + "defensive": (_blend(Q, low), dict(top_n=20, buffer=2.0, every=21)), + "lowrisk": (low, dict(top_n=20, buffer=2.0, every=21)), + } + + +def monthly_blend(R: pd.DataFrame, weights: dict[str, float]) -> pd.Series: + """Sleeves drift inside each month and are reset to ``weights`` at month + ends (the sleeves are separate accounts; only capital moves).""" + cols = list(weights) + w0 = np.array([weights[c] for c in cols]) + out = [] + for _, g in R[cols].groupby(R.index.to_period("M")): + grow = (1 + g).cumprod() + val = grow.mul(w0, axis=1).sum(axis=1) + prev = np.r_[1.0, val.values[:-1]] + out.append(pd.Series(val.values / prev - 1, index=g.index)) + return pd.concat(out) + + +def describe(r: pd.Series, blocks=BLOCKS) -> dict: + s = stats((1 + r).cumprod()) + bl = {f"{a[:4]}-{b[:4]}": stats((1 + r.loc[a:b]).cumprod())["sharpe"] + for a, b in blocks if len(r.loc[a:b]) > 200} + yrs = ((1 + r).groupby(r.index.year).prod() - 1) + return {**{k: float(v) for k, v in s.items()}, "blocks": bl, "yearly": {int(k): float(v) for k, v in yrs.items()}} + + +def line(name, d) -> str: + return (f"{name:44s} cagr {d['cagr']*100:5.1f}% sh {d['sharpe']:.2f} vol {d['vol']*100:4.1f}% " + f"mdd {d['mdd']*100:5.1f}% calmar {d['calmar']:.2f} worst yr {min(d['yearly'].values())*100:5.1f}% " + f"blocks [{' '.join(f'{v:.2f}' for v in d['blocks'].values())}]") + + +def main() -> None: + p, _ = load("us") + pf = p.slice() + pf.costs = Costs(p.costs.buy, p.costs.sell, None, 2.0, 10_000.0) # + $2 per order on a $10k sleeve + S = sleeve_scores(p) + R, Rfee = {}, {} + for name, (score, kw) in S.items(): + span = LONG if name in ("momentum", "lowrisk") else SPAN + R[name] = run(p, score.where(p.member), [span], **kw)["_rets"] + Rfee[name] = run(pf, score.where(p.member), [span], **kw)["_rets"] + ew = p.close.notna().astype(float).where(p.member) + R["ew_sp500"] = run(p, ew, [LONG], top_n=10000, every=10)["_rets"] + with open(os.path.join("data", "research", "tr_long.pkl"), "rb") as fh: + tr = pickle.load(fh)["equity"]["trend_rider_v7_vt24"] + R["trend"] = Rfee["trend"] = tr.pct_change().reindex(p.dates).loc["2010-03-01":] # $2/order already + R["spy"] = p.benchmark.ffill().pct_change().reindex(p.dates) + R = pd.DataFrame(R).loc[LONG[0]:LONG[1]] + Rfee = pd.DataFrame(Rfee).loc[LONG[0]:LONG[1]] + + out = {} + span = R.loc[SPAN[0]:SPAN[1], ["momentum", "defensive", "lowrisk", "trend", "spy"]].fillna(0.0) + monthly = (1 + span).resample("ME").prod() - 1 + print("monthly return correlation 2011-2026:\n", monthly.corr().round(2).to_string(), "\n") + print("=== sleeves (10 bp costs; trend includes $2/order)") + for c in R.columns: + r = R[c].dropna() + out[c] = describe(r) + print(line(f"{c} {r.index[0].year}-", out[c])) + three = {"momentum": 1 / 3, "defensive": 1 / 3, "trend": 1 / 3} + three_px = {"momentum": 1 / 3, "lowrisk": 1 / 3, "trend": 1 / 3} + print("\n=== portfolios 2011-2026, monthly rebalanced between sleeves") + for label, w, src in [("3 sleeves (defensive = quality+lowrisk)", three, R), + ("3 sleeves (defensive = lowrisk, price only)", three_px, R), + ("3 sleeves, + $2/order on $10k sleeves", three, Rfee), + ("momentum + defensive (no ETF sleeve)", {"momentum": .5, "defensive": .5}, R), + ("momentum + trend", {"momentum": .5, "trend": .5}, R)]: + r = monthly_blend(src.loc[SPAN[0]:SPAN[1]].fillna(0.0), w) + out[label] = describe(r) + print(line(label, out[label])) + print("\n=== price-only walk-back 2006-2026 (includes 2008; trend sleeve from 2010-03, before that 50/50)") + rr = R.loc[LONG[0]:LONG[1]] + pre = monthly_blend(rr.loc[:"2010-02-28", ["momentum", "lowrisk"]].fillna(0.0), {"momentum": .5, "lowrisk": .5}) + post = monthly_blend(rr.loc["2010-03-01":].fillna(0.0), three_px) + r = pd.concat([pre, post]) + out["walkback price-only"] = describe(r) + print(line("momentum + lowrisk (+ trend from 2010)", out["walkback price-only"])) + for c in ("momentum", "lowrisk", "ew_sp500", "spy"): + print(line(f" {c} 2006-", describe(rr[c].dropna()))) + + # deflated Sharpe: every configuration this and the previous round tried + n = 0 + for f in ("trials_us.csv", "trials_us_sweep.csv"): + path = os.path.join(RESULTS, f) + if os.path.exists(path): + with open(path) as fh: + n += sum(1 for line in fh if not line.startswith(("combo,", "score,"))) + n += 12 + 60 + 20 # rotation schemes, fund_eval variants, defensive variants + r = monthly_blend(R.loc[SPAN[0]:SPAN[1]].fillna(0.0), three) + dsr = deflated_sharpe(r, n) + out["dsr"] = {"n_trials": n, **{k: float(v) for k, v in dsr.items()}} + print(f"\nDSR of the 3-sleeve book: {dsr['dsr']:.3f} (Sharpe {dsr['sr_annual']:.2f} vs " + f"expected max of {n} unskilled trials {dsr['sr0_annual']:.2f})") + print("\nyearly returns (3 sleeves vs SPY):") + spy_y = out["spy"]["yearly"] + for y, v in out["3 sleeves (defensive = quality+lowrisk)"]["yearly"].items(): + print(f" {y}: {v*100:6.1f}% SPY {spy_y.get(y, float('nan'))*100:6.1f}%") + with open(os.path.join(RESULTS, "us_final.json"), "w") as fh: + json.dump(out, fh, indent=1) + R.to_pickle(os.path.join(RESULTS, "us_final_sleeves.pkl")) + + +if __name__ == "__main__": + main() diff --git a/research/lab/us_sweep.py b/research/lab/us_sweep.py new file mode 100644 index 0000000..e008996 --- /dev/null +++ b/research/lab/us_sweep.py @@ -0,0 +1,184 @@ +""" +US construction and overlay sweep for a few fixed scores, under one set of +rules (PIT S&P 500 members, next-open fills, 10 bp costs, phase averaging). + + PYTHONPATH=. uv run --with numba python -m research.lab.us_sweep --part construction + PYTHONPATH=. uv run --with numba python -m research.lab.us_sweep --part overlays + +Every configuration is appended to data/research/results/trials_us_sweep.csv +so the deflated Sharpe of the final pick can count them. +""" + +from __future__ import annotations + +import argparse +import os +import pickle +import warnings + +import numpy as np +import pandas as pd + +from research.lab import fundamentals as fu +from research.lab import overlays as ov +from research.lab.engine import Costs, simulate +from research.lab.factor_scan import RESULTS, load +from research.lab.report import stats +from research.lab.rotation import factor_ranks + +warnings.filterwarnings("ignore") + +# 2006-2026 includes the 2008 crash; fundamentals start in 2011 +LONG = [("2006-09-27", "2011-09-26"), ("2011-09-27", "2016-09-26"), + ("2016-09-27", "2021-09-26"), ("2021-09-27", "2026-09-24")] +SHORT = [("2011-01-01", "2014-12-31"), ("2015-01-01", "2018-12-31"), + ("2019-01-01", "2022-12-31"), ("2023-01-01", "2026-09-24")] + + +def _blend(*dfs: pd.DataFrame) -> pd.DataFrame: + stack = np.stack([d.values for d in dfs]) + with np.errstate(invalid="ignore"): + m = np.nanmean(stack, axis=0) + return pd.DataFrame(m, index=dfs[0].index, columns=dfs[0].columns) + + +def sector_rank(df: pd.DataFrame, p) -> pd.DataFrame: + df = df.where(p.member) + sec = p.extra["sector"] + out = pd.DataFrame(np.nan, index=df.index, columns=df.columns) + for s in sec.dropna().unique(): + cols = [c for c in sec.index[sec == s] if c in df.columns] + out[cols] = df[cols].rank(axis=1, pct=True) + return out + + +def scores(p) -> dict[str, tuple[pd.DataFrame, list]]: + pr = factor_ranks(p, "us") + F = fu.load(p) + Q = _blend(*(sector_rank(F[k], p) for k in ("gp_a", "cfo_a", "roe", "accruals", "leverage"))) + mom = _blend(pr["mom_12_1"], pr["resid_mom"]) + low = _blend(pr["ivol"], pr["low_beta"]) + return { + "champ": (_blend(pr["overnight_mom"], pr["rec_126"]).where(p.member), LONG), + "mom_qual_low": (_blend(mom, Q, low).where(p.member), SHORT), + } + + +def run(p, score, subs, exposure=None, **kw) -> dict: + """Phase-averaged daily returns over the whole span plus block stats.""" + a, b = subs[0][0], subs[-1][1] + every = kw.get("every", 10) + w0 = p.dates.searchsorted(pd.Timestamp(a)) + sp = p.slice(p.dates[max(w0 - 5, 0)], b) + rets = [] + for off in range(every): + r = simulate(sp, score, offset=off, warmup=5, exposure=exposure, **kw) + rets.append(r.returns.loc[a:b]) + if off == 0: + turn = r.turnover.loc[a:b].sum() / ((pd.Timestamp(b) - pd.Timestamp(a)).days / 365.25) + R = pd.concat(rets, axis=1) + per_phase = [stats((1 + R[c]).cumprod()) for c in R.columns] + avg = R.mean(axis=1) # = a book split into ``every`` tranches + out = {"sharpe": np.mean([s["sharpe"] for s in per_phase]), + "cagr": np.mean([s["cagr"] for s in per_phase]), + "mdd": np.mean([s["mdd"] for s in per_phase]), + "sharpe_min_phase": min(s["sharpe"] for s in per_phase), + "tranche_sharpe": stats((1 + avg).cumprod())["sharpe"], "turn": turn, + "_rets": avg} + blocks = [] + for sa, sb in subs: + blocks.append(np.mean([stats((1 + R[c].loc[sa:sb]).cumprod())["sharpe"] for c in R.columns])) + out["blocks"] = blocks + out["min_block"] = min(blocks) + return out + + +def fmt(name, st) -> str: + return (f"{name:48s} sh {st['sharpe']:.2f} [{st['sharpe_min_phase']:.2f}] tr {st['tranche_sharpe']:.2f} " + f"cagr {st['cagr']*100:5.1f}% mdd {st['mdd']*100:5.1f}% turn {st['turn']:4.1f}x " + f"min {st['min_block']:.2f} [{' '.join(f'{x:.2f}' for x in st['blocks'])}]") + + +def log_trials(rows, part): + os.makedirs(RESULTS, exist_ok=True) + path = os.path.join(RESULTS, "trials_us_sweep.csv") + df = pd.DataFrame([{k: v for k, v in r.items() if not k.startswith("_") and k != "blocks"} for r in rows]) + df["part"] = part + df.to_csv(path, mode="a", header=not os.path.exists(path), index=False) + + +def construction(p, S) -> list[dict]: + rows = [] + grid = [] + for top in (10, 20, 30): + for buf in (0.0, 1.0, 2.0): + grid.append((f"top{top} buf{buf:g} equal", dict(top_n=top, buffer=buf, weighting="equal"))) + for top in (10, 20, 30): + grid.append((f"top{top} buf1 zvol", dict(top_n=top, buffer=1.0, weighting="zvol"))) + grid.append((f"top{top} buf1 invvol", dict(top_n=top, buffer=1.0, weighting="invvol"))) + grid.append(("dyn 10-30 thr0.80 buf1 equal", dict(top_n=30, buffer=1.0, dyn_n=(10, 0.80)))) + grid.append(("dyn 10-30 thr0.75 buf1 equal", dict(top_n=30, buffer=1.0, dyn_n=(10, 0.75)))) + grid.append(("top20 buf1 every5", dict(top_n=20, buffer=1.0, every=5))) + grid.append(("top20 buf1 every21", dict(top_n=20, buffer=1.0, every=21))) + for sname, (score, subs) in S.items(): + print(f"=== {sname}", flush=True) + for label, kw in grid: + kw = {"every": 10, **kw} + st = run(p, score, subs, **kw) + rows.append({"score": sname, "config": label, **st}) + print(fmt(label, st), flush=True) + return rows + + +def overlays(p, S, base_kw) -> list[dict]: + lvl = p.benchmark.ffill() + ew = ov.ew_index(p) + X = { + "none": None, + "spy derisk_vol": ov.derisk_vol(lvl), + "spy faber10m": ov.faber_sma(lvl), + "spy jump_model": ov.jump_model(lvl, p.cash_ret), + "spy dm_state24m": ov.momentum_state(lvl), + "ew breadth-free derisk_vol": ov.derisk_vol(ew), + } + rows = [] + for sname, (score, subs) in S.items(): + print(f"=== {sname} {base_kw}", flush=True) + base = run(p, score, subs, **base_kw) + # strategy-level de-risk-only vol target from its own trailing returns + own = base["_rets"] + own_lvl = (1 + own).cumprod() + X_s = dict(X) + X_s["own derisk_vol"] = ov.derisk_vol(own_lvl, min_hist=126).reindex(p.dates).fillna(1.0) + for name, e in X_s.items(): + st = base if e is None else run(p, score, subs, exposure=e, **base_kw) + rows.append({"score": sname, "config": f"overlay {name}", **st}) + print(fmt(f"overlay {name}", st), flush=True) + return rows + + +def main() -> None: + ap = argparse.ArgumentParser() + ap.add_argument("--part", default="construction") + ap.add_argument("--top-n", type=int, default=20) + ap.add_argument("--buffer", type=float, default=1.0) + ap.add_argument("--weighting", default="equal") + ap.add_argument("--fee", type=float, default=0.0, help="fixed fee per order on $10k") + args = ap.parse_args() + p, _ = load("us") + if args.fee: + p = p.slice() + p.costs = Costs(p.costs.buy, p.costs.sell, None, args.fee, 10_000.0) + S = scores(p) + if args.part == "construction": + rows = construction(p, S) + else: + rows = overlays(p, S, dict(top_n=args.top_n, buffer=args.buffer, every=10, + weighting=args.weighting)) + log_trials(rows, args.part) + with open(os.path.join(RESULTS, f"us_sweep_{args.part}.pkl"), "wb") as fh: + pickle.dump(rows, fh) + + +if __name__ == "__main__": + main() diff --git a/strategies/robust_momentum.py b/strategies/robust_momentum.py index e93d1dc..69fba97 100644 --- a/strategies/robust_momentum.py +++ b/strategies/robust_momentum.py @@ -17,6 +17,10 @@ Signals: lowrisk_momentum (CN) amplitude-cut momentum + low beta + negative price-volume correlation + momentum/vol lowbeta_sharpe_mom (CN) low beta + momentum/vol (close prices only) + low_risk (US) low idiosyncratic volatility + low beta (close only); + the defensive sleeve of the 2026-09-29 US portfolio + research, meant to be held next to overnight_recovery + and trend_rider_v7_vt24 (monthly correlation 0.2-0.4) """ import numpy as np @@ -68,6 +72,17 @@ def low_beta(close: pd.DataFrame, window: int = 252) -> pd.DataFrame: return -(e_xy - e_x.mul(e_y, axis=0)).div(var_y, axis=0) +def idiosyncratic_vol(close: pd.DataFrame, beta_window: int = 252, + vol_window: int = 63) -> pd.DataFrame: + """Negative 63-day volatility of returns net of the equal-weight market + times a trailing (lagged) 252-day beta; Ang, Hodrick, Xing & Zhang (2006).""" + r = _returns(close) + m = r.mean(axis=1).fillna(0.0) + beta = -low_beta(close, beta_window).shift(1) + resid = r - beta.mul(m, axis=0) + return -resid.rolling(vol_window, min_periods=40).std() + + def price_volume_corr(close: pd.DataFrame, volume: pd.DataFrame) -> pd.DataFrame: """Negative 21-day correlation of price and dollar volume.""" return -close.ffill().rolling(21, min_periods=15).corr(close * volume) @@ -115,11 +130,17 @@ def signal_lowbeta_sharpe_mom(bars): return _composite(low_beta(c), sharpe_momentum(c)) +def signal_low_risk(bars): + c = bars["close"] + return _composite(idiosyncratic_vol(c), low_beta(c)) + + # name -> (bar fields, score function) SIGNALS = { "overnight_recovery": (("close", "open"), signal_overnight_recovery), "lowrisk_momentum": (("close", "high", "low", "volume"), signal_lowrisk_momentum), "lowbeta_sharpe_mom": (("close",), signal_lowbeta_sharpe_mom), + "low_risk": (("close",), signal_low_risk), } diff --git a/strategies/sleeve_portfolio.py b/strategies/sleeve_portfolio.py new file mode 100644 index 0000000..b26b5b4 --- /dev/null +++ b/strategies/sleeve_portfolio.py @@ -0,0 +1,140 @@ +""" +Sleeve portfolio: several strategies held side by side in one account, each +with a fixed share of capital. + +``us_three_sleeve`` is the portfolio from the 2026-09-29 US research +(research/lab/us_final.py, report https://claude.ai/artifact/VGhZvKkS9UBYZkLeqGjcNC): +one third each of + + momentum : overnight_recovery_top10 (S&P 500 stocks) + defensive : lowrisk_top20 (S&P 500 stocks) + trend : trend_rider_v7_vt24 (leveraged / defensive ETFs) + +Their monthly return correlations are 0.2-0.4, which is where the portfolio's +higher Sharpe and smaller drawdown come from (2011-2026: Sharpe 1.21 vs +0.88-1.03 for the sleeves alone). + +Each sleeve keeps its own strategy's targets. The stock sleeves see only the +stock columns and the trend sleeve only its ETFs. A sleeve's share of the +account drifts with its own return, like a separate account would, and all +shares are reset to the fixed ones at the first session of a month when any +share is more than ``band`` (relative) away from its fixed value. Resetting +every month instead would resize most of the ~30 names each month, which the +live book pays a ticket fee for. + +Timing: every sleeve decides on completed bars, so the trend sleeve reads +closes through the previous session here, where the stand-alone +trend_rider_v7_vt24 reads the session's open as its latest price. +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd + +from strategies.base import BarStrategy, Strategy + + +def decisions(strategy: Strategy, bars: dict[str, pd.DataFrame]) -> pd.DataFrame: + """Targets decided on each row's own bars. + + A Strategy's weights on row t use data through t-1, so the decision made + on the last row only appears on a row after it. Padding one copy of the + last row and shifting back yields it; a causal strategy ignores the value + of the padded row.""" + close = bars["close"] + nxt = close.index[-1] + pd.Timedelta(days=1) + padded = {f: pd.concat([b, b.iloc[[-1]].set_axis([nxt], axis=0)]) for f, b in bars.items()} + data = padded if isinstance(strategy, BarStrategy) else padded["close"] + weights = strategy.generate_signals(data) + weights = weights.reindex(index=padded["close"].index, columns=close.columns).fillna(0.0) + return weights.shift(-1).iloc[:-1].set_axis(close.index, axis=0) + + +def drifting_shares(sleeve_weights: list[np.ndarray], returns: np.ndarray, base: np.ndarray, + month_start: np.ndarray, band: float) -> np.ndarray: + """Share of the account in each sleeve per row (rows x sleeves). + + Row k's shares grow with each sleeve's return from k-1 to k (its row k-1 + weights times row k asset returns; the unallocated part of a sleeve is + cash) and are reset to ``base`` on a month-start row when any share is + more than ``band`` away from its base, relatively.""" + n = returns.shape[0] + shares = np.empty((n, len(base))) + s = base.astype(float).copy() + for k in range(n): + if k > 0: + growth = np.array([1.0 + w[k - 1] @ returns[k] for w in sleeve_weights]) + s = s * growth + s = s / s.sum() if s.sum() > 0 else base.copy() + if month_start[k] and np.max(np.abs(s / base - 1.0)) > band: + s = base.astype(float).copy() + shares[k] = s + return shares + + +class SleevePortfolioStrategy(BarStrategy): + """Fixed-share portfolio of sleeve strategies in one account. + + ``sleeves`` is a list of ``(name, strategy, share, universe)``; universe + "stocks" gives the sleeve every column except ``etfs``, "etfs" gives it + the ``etfs`` columns present.""" + + def __init__(self, sleeves: list[tuple[str, Strategy, float, str]], etfs: list[str], + band: float = 0.25): + total = sum(share for _, _, share, _ in sleeves) + self.sleeves = [(n, s, share / total, u) for n, s, share, u in sleeves] + self.etfs = list(etfs) + self.band = band + fields = {"close"} + for _, s, _, _ in sleeves: + fields |= set(s.fields) if isinstance(s, BarStrategy) else set() + self.fields = tuple(sorted(fields, key=lambda f: (f != "close", f))) + + def _sleeve_bars(self, strategy: Strategy, bars: dict, cols: list[str]) -> dict: + wanted = strategy.fields if isinstance(strategy, BarStrategy) else ("close",) + sub = {f: bars[f][cols] for f in wanted} + if "eligible" in bars: + sub["eligible"] = bars["eligible"].reindex(columns=cols).fillna(True) + return sub + + def target_weights(self, bars: dict[str, pd.DataFrame]) -> tuple[pd.DataFrame, pd.DataFrame]: + """Unshifted combined targets (row k decided on bars through k) and the + sleeve shares behind them.""" + close = bars["close"] + etf_cols = [c for c in close.columns if c in self.etfs] + stock_cols = [c for c in close.columns if c not in self.etfs] + parts = [] + for _, strategy, _, universe in self.sleeves: + cols = etf_cols if universe == "etfs" else stock_cols + d = decisions(strategy, self._sleeve_bars(strategy, bars, cols)) + parts.append(d.reindex(columns=close.columns).fillna(0.0).to_numpy()) + rets = np.nan_to_num(close.ffill().pct_change(fill_method=None).to_numpy(dtype=float), nan=0.0) + months = close.index.to_period("M") + month_start = np.r_[False, months[1:] != months[:-1]] + base = np.array([share for _, _, share, _ in self.sleeves]) + shares = drifting_shares(parts, rets, base, month_start, self.band) + combined = sum(shares[:, [j]] * parts[j] for j in range(len(parts))) + names = [n for n, _, _, _ in self.sleeves] + return (pd.DataFrame(combined, index=close.index, columns=close.columns), + pd.DataFrame(shares, index=close.index, columns=names)) + + def generate_weights(self, bars: dict[str, pd.DataFrame]) -> pd.DataFrame: + combined, _ = self.target_weights(bars) + return combined.shift(1).fillna(0.0) + + +def us_three_sleeve() -> SleevePortfolioStrategy: + from strategies.permanent import ETF_UNIVERSE + from strategies.robust_momentum import RobustMomentumStrategy + from strategies.trend_rider_v7 import TrendRiderV7 + return SleevePortfolioStrategy( + [ + ("momentum", RobustMomentumStrategy("overnight_recovery", top_n=10, rebal_freq=10, + buffer=2.0), 1.0, "stocks"), + ("defensive", RobustMomentumStrategy("low_risk", top_n=20, rebal_freq=21, + buffer=2.0), 1.0, "stocks"), + ("trend", TrendRiderV7(target_vol=0.24, min_lev=0.5), 1.0, "etfs"), + ], + etfs=sorted(set(ETF_UNIVERSE)), + ) diff --git a/tests/test_sleeve_portfolio.py b/tests/test_sleeve_portfolio.py new file mode 100644 index 0000000..557d3a5 --- /dev/null +++ b/tests/test_sleeve_portfolio.py @@ -0,0 +1,93 @@ +"""Contracts of the sleeve portfolio (us_three_sleeve): causal targets, sleeves +kept to their own columns, shares that drift like separate accounts, and the +live trader's session timing.""" + +import unittest + +import numpy as np +import pandas as pd + +import trader +from strategies.permanent import ETF_UNIVERSE +from strategies.robust_momentum import RobustMomentumStrategy +from strategies.sleeve_portfolio import (SleevePortfolioStrategy, decisions, drifting_shares, + us_three_sleeve) + + +def _bars(n_days: int = 520, n_stocks: int = 30, seed: int = 1) -> dict[str, pd.DataFrame]: + rng = np.random.default_rng(seed) + idx = pd.bdate_range("2024-01-01", periods=n_days) + cols = [f"S{i:02d}" for i in range(n_stocks)] + sorted(set(ETF_UNIVERSE)) + r = rng.normal(0.0005, 0.015, (n_days, len(cols))) + close = pd.DataFrame(100 * np.exp(np.cumsum(r, axis=0)), index=idx, columns=cols) + opens = close.shift(1).bfill() * np.exp(rng.normal(0, 0.004, close.shape)) + return {"close": close, "open": opens} + + +class SleevePortfolioTests(unittest.TestCase): + def test_targets_on_a_day_ignore_that_day_and_later_bars(self): + bars = _bars() + strat = us_three_sleeve() + cut = bars["close"].index[450] + base = strat.generate_signals(bars) + shocked = {f: b.copy() for f, b in bars.items()} + for f in shocked: + shocked[f].loc[cut:] *= 1.7 + again = strat.generate_signals(shocked) + pd.testing.assert_frame_equal(base.loc[:cut], again.loc[:cut]) + + def test_stock_sleeves_hold_no_etfs_and_the_book_is_not_levered(self): + bars = _bars() + strat = us_three_sleeve() + weights = strat.generate_signals(bars) + etfs = set(ETF_UNIVERSE) + self.assertLessEqual(weights.sum(axis=1).max(), 1.0 + 1e-9) + # After warm-up the stock part is two sleeves' worth of whole names only. + late = weights.iloc[-1] + stocks = late[[c for c in late.index if c not in etfs]] + self.assertEqual(int((stocks > 0).sum()) <= 30, True) + self.assertGreater(stocks.sum(), 0.5) + _, shares = strat.target_weights(bars) + self.assertTrue(np.allclose(shares.sum(axis=1), 1.0)) + + def test_decisions_are_the_strategy_weights_one_row_earlier(self): + bars = _bars() + strat = RobustMomentumStrategy("low_risk", top_n=5, rebal_freq=5, buffer=1.0) + stock = {f: b[[c for c in b.columns if c.startswith("S")]] for f, b in bars.items()} + d = decisions(strat, stock) + w = strat.generate_signals(stock) + pd.testing.assert_frame_equal(d.iloc[:-1], w.shift(-1).iloc[:-1].fillna(0.0)) + self.assertAlmostEqual(d.iloc[-1].sum(), 1.0) + + def test_shares_drift_with_sleeve_returns_and_reset_past_the_band(self): + # Two sleeves fully in one asset each; A doubles on day 1, B is flat. + returns = np.array([[0.0, 0.0], [1.0, 0.0], [0.0, 0.0], [0.0, 0.0]]) + wa = np.array([[1.0, 0.0]] * 4) + wb = np.array([[0.0, 1.0]] * 4) + base = np.array([0.5, 0.5]) + month_start = np.array([False, False, False, True]) + shares = drifting_shares([wa, wb], returns, base, month_start, band=0.25) + np.testing.assert_allclose(shares[1], [2 / 3, 1 / 3]) # drifted like two accounts + np.testing.assert_allclose(shares[2], [2 / 3, 1 / 3]) # no reset mid-month + np.testing.assert_allclose(shares[3], [0.5, 0.5]) # 33% off base: reset + kept = drifting_shares([wa, wb], returns, base, month_start, band=0.5) + np.testing.assert_allclose(kept[3], [2 / 3, 1 / 3]) # inside a 50% band + + def test_trader_session_targets_use_bars_through_the_previous_day(self): + bars = _bars() + strat = us_three_sleeve() + days = bars["close"].index[-2:] + full = strat.generate_signals(bars).loc[days] + got = trader.bar_session_weights(strat, bars, days) + pd.testing.assert_frame_equal(got, full) + + def test_registered_as_a_us_mixed_strategy(self): + self.assertIsInstance(trader.STRATEGY_REGISTRY["us_three_sleeve"](), SleevePortfolioStrategy) + self.assertEqual(set(trader.strategy_extra_tickers("us_three_sleeve")), set(ETF_UNIVERSE)) + self.assertTrue(trader.strategy_supports_market("us_three_sleeve", "us")) + self.assertFalse(trader.strategy_supports_market("us_three_sleeve", "cn")) + self.assertEqual(trader.strategy_fields("us_three_sleeve"), ("close", "open")) + + +if __name__ == "__main__": + unittest.main() diff --git a/trader.py b/trader.py index f58f36d..3985b71 100644 --- a/trader.py +++ b/trader.py @@ -58,6 +58,7 @@ from strategies.permanent import ( ) from strategies.recovery_momentum import RecoveryMomentumStrategy from strategies.robust_momentum import RobustMomentumStrategy +from strategies.sleeve_portfolio import us_three_sleeve from strategies.trend_following import TrendFollowingStrategy from strategies.trend_rider_v5 import TrendRiderV5 from strategies.trend_rider_voltgt import ( @@ -290,6 +291,13 @@ STRATEGY_REGISTRY = { "lowbeta_sharpe_mom_top20": lambda **kw: RobustMomentumStrategy( "lowbeta_sharpe_mom", top_n=20, rebal_freq=10, buffer=1.0, ), + # --- US portfolio research (2026-09-29): defensive sleeve and the 3-sleeve book --- + "lowrisk_top20": lambda **kw: RobustMomentumStrategy( + "low_risk", top_n=20, rebal_freq=21, buffer=2.0, + ), + # 1/3 each of overnight_recovery_top10, lowrisk_top20 and trend_rider_v7_vt24 + # in one account (strategies/sleeve_portfolio.py) + "us_three_sleeve": lambda **kw: us_three_sleeve(), # --- TrendRiderV6: stock-picking + V5 regime engine --- "trend_rider_v6": lambda **kw: TrendRiderV6(), "trend_rider_v6_top10": lambda **kw: TrendRiderV6(top_n=10), @@ -319,6 +327,7 @@ ETF_STRATEGY_UNIVERSES = { MIXED_STRATEGY_EXTRA_TICKERS = { "trend_rider_v6": sorted(set(ETF_UNIVERSE)), "trend_rider_v6_top10": sorted(set(ETF_UNIVERSE)), + "us_three_sleeve": sorted(set(ETF_UNIVERSE)), } # Registered strategies the live monitor does not run, with the reason.