Add weekly strategy report comparing top strategies to baselines

Makefile

@@ -0,0 +1,4 @@
+.PHONY: rank
+
+rank:
+	uv run python rank_strategies.py $(ARGS)

rank_strategies.py

@@ -0,0 +1,159 @@
+#!/usr/bin/env python3
+"""Rank live strategy state files by current marked-to-market return."""
+
+import argparse
+import glob
+import json
+import os
+from dataclasses import dataclass
+from pathlib import Path
+
+import data_manager
+from trader import get_prices_for_date, portfolio_value
+from universe import UNIVERSES
+
+
+@dataclass
+class RankedStrategy:
+    strategy: str
+    return_pct: float
+    value: float
+    cash: float
+    n_positions: int
+    buys: int
+    sells: int
+    holdings: str
+    state_date: str
+
+
+def _state_name(path: str, market: str) -> str:
+    base = os.path.basename(path)
+    prefix = f"trader_{market}_"
+    return base[len(prefix):-len(".json")]
+
+
+def _load_close_data(market: str, update: bool):
+    if update:
+        universe = UNIVERSES[market]
+        tickers = universe["fetch"]()
+        benchmark = universe["benchmark"]
+        all_tickers = sorted(set(tickers + [benchmark]))
+        return data_manager.update(market, all_tickers)
+
+    data = data_manager.load(market)
+    if data is None:
+        raise RuntimeError(
+            f"No cached data found for market '{market}'. Run without --no-update first."
+        )
+    return data
+
+
+def _format_holdings(holdings: dict) -> str:
+    return ", ".join(f"{ticker}:{shares:g}" for ticker, shares in sorted(holdings.items()))
+
+
+def rank_market(market: str, update: bool, include_sim: bool) -> tuple[str, list[RankedStrategy]]:
+    close = _load_close_data(market, update)
+    latest_date = str(close.index[-1].date())
+
+    rows: list[RankedStrategy] = []
+    for path in sorted(glob.glob(f"data/trader_{market}_*.json")):
+        strategy = _state_name(path, market)
+        if strategy.startswith("sim_") and not include_sim:
+            continue
+
+        state = json.loads(Path(path).read_text())
+        if "initial_capital" not in state:
+            continue
+
+        holdings = state.get("holdings", {}) or {}
+        cash = float(state.get("cash", 0.0) or 0.0)
+        prices = get_prices_for_date(list(holdings), close.index[-1], close)
+        value = portfolio_value(holdings, prices, cash)
+        initial = float(state["initial_capital"])
+        return_pct = (value / initial - 1.0) * 100.0 if initial else 0.0
+
+        trades = state.get("trade_log", []) or []
+        buys = sum(1 for trade in trades if trade.get("action") == "BUY")
+        sells = sum(1 for trade in trades if trade.get("action") == "SELL")
+        equity = state.get("daily_equity", {}) or {}
+        state_date = max(equity.keys()) if equity else ""
+
+        rows.append(
+            RankedStrategy(
+                strategy=strategy,
+                return_pct=return_pct,
+                value=value,
+                cash=cash,
+                n_positions=len(holdings),
+                buys=buys,
+                sells=sells,
+                holdings=_format_holdings(holdings),
+                state_date=state_date,
+            )
+        )
+
+    rows.sort(key=lambda row: row.return_pct, reverse=True)
+    return latest_date, rows
+
+
+def print_market_table(market: str, latest_date: str, rows: list[RankedStrategy], top: int) -> None:
+    print(f"\n{market.upper()} Top{top}  latest_price_date={latest_date}")
+    print(
+        f"{'#':>2}  {'Strategy':<42} {'Return':>8} {'Value':>10} "
+        f"{'Cash':>9} {'Pos':>3} {'Buy/Sell':>8} {'StateDate':>10}  Holdings"
+    )
+    print("-" * 132)
+
+    for idx, row in enumerate(rows[:top], 1):
+        print(
+            f"{idx:>2}  {row.strategy:<42} {row.return_pct:>7.2f}% "
+            f"{row.value:>10.2f} {row.cash:>9.2f} {row.n_positions:>3} "
+            f"{row.buys:>3}/{row.sells:<4} {row.state_date:>10}  {row.holdings}"
+        )
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Update market data and rank live strategy state files."
+    )
+    parser.add_argument(
+        "--market",
+        choices=["all", "us", "cn"],
+        default="all",
+        help="Market to rank. Default: all.",
+    )
+    parser.add_argument(
+        "--top",
+        type=int,
+        default=10,
+        help="Number of strategies to print per market. Default: 10.",
+    )
+    parser.add_argument(
+        "--no-update",
+        action="store_true",
+        help="Use cached data only; do not download new prices.",
+    )
+    parser.add_argument(
+        "--include-sim",
+        action="store_true",
+        help="Include state files whose strategy name starts with sim_.",
+    )
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    markets = ["us", "cn"] if args.market == "all" else [args.market]
+
+    for market in markets:
+        latest_date, rows = rank_market(
+            market=market,
+            update=not args.no_update,
+            include_sim=args.include_sim,
+        )
+        print_market_table(market, latest_date, rows, args.top)
+
+
+if __name__ == "__main__":
+    main()

weekly_strategy_report.py

@@ -0,0 +1,309 @@
+#!/usr/bin/env python3
+"""Weekly top-strategy report against market baselines."""
+
+import argparse
+import glob
+import json
+import time
+from dataclasses import dataclass
+from datetime import timedelta
+from pathlib import Path
+from urllib.parse import urlencode
+from urllib.request import Request, urlopen
+
+import pandas as pd
+import yfinance as yf
+
+import data_manager
+
+INITIAL_VALUE = 10_000.0
+
+
+@dataclass(frozen=True)
+class Baseline:
+    label: str
+    yahoo_symbol: str | None = None
+    cache_symbol: str | None = None
+    sohu_code: str | None = None
+
+
+@dataclass
+class StrategyEquity:
+    name: str
+    series: pd.Series
+    total_return: float
+
+
+BASELINES = {
+    "us": [
+        Baseline("NASDAQ Composite", yahoo_symbol="^IXIC"),
+        Baseline("SPY", yahoo_symbol="SPY", cache_symbol="SPY"),
+    ],
+    "cn": [
+        Baseline("CSI 300", yahoo_symbol="000300.SS", cache_symbol="000300.SS", sohu_code="zs_000300"),
+        Baseline("CSI 800", yahoo_symbol="000906.SS", sohu_code="zs_000906"),
+    ],
+}
+
+
+def _state_name(path: str, market: str) -> str:
+    base = Path(path).name
+    prefix = f"trader_{market}_"
+    return base[len(prefix):-len(".json")]
+
+
+def load_strategies(market: str, include_sim: bool) -> list[StrategyEquity]:
+    rows: list[StrategyEquity] = []
+    for path in sorted(glob.glob(f"data/trader_{market}_*.json")):
+        name = _state_name(path, market)
+        if name.startswith("sim_") and not include_sim:
+            continue
+
+        state = json.loads(Path(path).read_text())
+        daily_equity = state.get("daily_equity", {}) or {}
+        if len(daily_equity) < 2:
+            continue
+
+        series = pd.Series(daily_equity, dtype=float)
+        series.index = pd.to_datetime(series.index)
+        series = series.sort_index()
+
+        initial = float(state.get("initial_capital") or series.iloc[0] or INITIAL_VALUE)
+        total_return = series.iloc[-1] / initial - 1.0 if initial else 0.0
+        rows.append(StrategyEquity(name=name, series=series, total_return=total_return))
+
+    rows.sort(key=lambda row: row.total_return, reverse=True)
+    return rows
+
+
+def _close_from_yahoo_frame(raw: pd.DataFrame, symbol: str) -> pd.Series:
+    if raw.empty:
+        return pd.Series(dtype=float)
+
+    if isinstance(raw.columns, pd.MultiIndex):
+        if "Close" not in raw.columns.get_level_values(0):
+            return pd.Series(dtype=float)
+        close = raw["Close"]
+        if isinstance(close, pd.DataFrame):
+            if symbol in close.columns:
+                return close[symbol].dropna().astype(float)
+            if len(close.columns) == 1:
+                return close.iloc[:, 0].dropna().astype(float)
+        return close.dropna().astype(float)
+
+    if "Close" not in raw.columns:
+        return pd.Series(dtype=float)
+    return raw["Close"].dropna().astype(float)
+
+
+def download_yahoo_close(symbol: str, start: pd.Timestamp, end: pd.Timestamp) -> pd.Series:
+    end_exclusive = (end + timedelta(days=1)).strftime("%Y-%m-%d")
+    raw = yf.download(
+        symbol,
+        start=start.strftime("%Y-%m-%d"),
+        end=end_exclusive,
+        auto_adjust=True,
+        progress=False,
+    )
+    series = _close_from_yahoo_frame(raw, symbol)
+    series.index = pd.to_datetime(series.index).tz_localize(None)
+    return series
+
+
+def download_sohu_close(code: str, start: pd.Timestamp, end: pd.Timestamp) -> pd.Series:
+    params = {
+        "code": code,
+        "start": start.strftime("%Y%m%d"),
+        "end": end.strftime("%Y%m%d"),
+        "stat": "1",
+        "order": "D",
+        "period": "d",
+        "callback": "historySearchHandler",
+        "rt": "jsonp",
+    }
+    url = "https://q.stock.sohu.com/hisHq?" + urlencode(params)
+    req = Request(url, headers={"User-Agent": "Mozilla/5.0"})
+    text = ""
+    for attempt in range(3):
+        try:
+            with urlopen(req, timeout=20) as resp:
+                text = resp.read().decode("gbk").strip()
+            break
+        except Exception:
+            if attempt == 2:
+                return pd.Series(dtype=float)
+            time.sleep(1)
+
+    prefix = "historySearchHandler("
+    if not text.startswith(prefix) or not text.endswith(")"):
+        return pd.Series(dtype=float)
+
+    payload = json.loads(text[len(prefix):-1])
+    if not payload or payload[0].get("status") != 0:
+        return pd.Series(dtype=float)
+
+    rows = payload[0].get("hq", [])
+    if not rows:
+        return pd.Series(dtype=float)
+
+    data = {row[0]: float(row[2]) for row in rows if len(row) >= 3}
+    series = pd.Series(data, dtype=float)
+    series.index = pd.to_datetime(series.index)
+    return series.sort_index()
+
+
+def clip_dates(series: pd.Series, start: pd.Timestamp, end: pd.Timestamp) -> pd.Series:
+    if series.empty:
+        return series
+    clipped = series.copy()
+    clipped.index = pd.to_datetime(clipped.index).tz_localize(None)
+    return clipped.loc[(clipped.index >= start) & (clipped.index <= end)].dropna()
+
+
+def load_baseline(market: str, baseline: Baseline, start: pd.Timestamp, end: pd.Timestamp) -> pd.Series:
+    if baseline.sohu_code:
+        series = download_sohu_close(baseline.sohu_code, start, end)
+    else:
+        series = pd.Series(dtype=float)
+
+    series = clip_dates(series, start, end)
+    if len(series) < 2:
+        cached = data_manager.load(market)
+        if cached is not None and baseline.cache_symbol in cached.columns:
+            series = cached[baseline.cache_symbol].dropna().astype(float)
+        else:
+            series = pd.Series(dtype=float)
+
+    series = clip_dates(series, start, end)
+    if len(series) < 2 and baseline.yahoo_symbol:
+        series = download_yahoo_close(baseline.yahoo_symbol, start, end)
+
+    series = clip_dates(series, start, end)
+    if len(series) < 2 and baseline.sohu_code:
+        series = download_sohu_close(baseline.sohu_code, start, end)
+
+    series = clip_dates(series, start, end)
+    if series.empty:
+        return series
+    return series / series.iloc[0] * INITIAL_VALUE
+
+
+def select_period(strategies: list[StrategyEquity], start: str | None, end: str | None) -> tuple[pd.Timestamp, pd.Timestamp]:
+    if start:
+        start_ts = pd.Timestamp(start)
+    else:
+        start_ts = max(row.series.index.min() for row in strategies)
+
+    if end:
+        end_ts = pd.Timestamp(end)
+    else:
+        end_ts = min(row.series.index.max() for row in strategies)
+
+    if start_ts > end_ts:
+        raise ValueError(f"Invalid period: start {start_ts.date()} is after end {end_ts.date()}")
+    return start_ts, end_ts
+
+
+def weekly_last(frame: pd.DataFrame) -> pd.DataFrame:
+    rows = []
+    for _, group in frame.groupby(pd.Grouper(freq="W-FRI")):
+        group = group.dropna(how="all")
+        if group.empty:
+            continue
+        last = group.iloc[-1].copy()
+        last.name = group.index[-1].strftime("%Y-%m-%d")
+        rows.append(last)
+    if not rows:
+        return pd.DataFrame(columns=frame.columns)
+    result = pd.DataFrame(rows)
+    result.index.name = "week_date"
+    return result
+
+
+def build_market_report(
+    market: str,
+    top: int,
+    include_sim: bool,
+    start: str | None,
+    end: str | None,
+) -> tuple[pd.DataFrame, pd.DataFrame, list[str]]:
+    strategies = load_strategies(market, include_sim=include_sim)
+    if not strategies:
+        raise RuntimeError(f"No strategy equity data found for market '{market}'")
+
+    selected = strategies[:top]
+    start_ts, end_ts = select_period(selected, start, end)
+
+    frame = pd.DataFrame({
+        row.name: row.series.loc[(row.series.index >= start_ts) & (row.series.index <= end_ts)]
+        for row in selected
+    })
+
+    warnings = []
+    for baseline in BASELINES[market]:
+        series = load_baseline(market, baseline, start_ts, end_ts)
+        if series.empty:
+            warnings.append(f"{market.upper()} baseline '{baseline.label}' has no data for {start_ts.date()} to {end_ts.date()}")
+            continue
+        frame[baseline.label] = series
+
+    weekly_values = weekly_last(frame)
+    weekly_returns = (weekly_values / INITIAL_VALUE - 1.0) * 100.0
+    return weekly_values, weekly_returns, warnings
+
+
+def write_outputs(market: str, values: pd.DataFrame, returns: pd.DataFrame, output_dir: Path) -> tuple[Path, Path]:
+    output_dir.mkdir(parents=True, exist_ok=True)
+    value_path = output_dir / f"weekly_{market}_top10_vs_baselines.csv"
+    return_path = output_dir / f"weekly_{market}_top10_vs_baselines_returns.csv"
+    values.round(2).to_csv(value_path)
+    returns.round(4).to_csv(return_path)
+    return value_path, return_path
+
+
+def print_returns(market: str, returns: pd.DataFrame) -> None:
+    print(f"\n{market.upper()} weekly return %")
+    if returns.empty:
+        print("  No weekly rows.")
+        return
+    printable = returns.copy()
+    printable = printable.map(lambda value: "" if pd.isna(value) else f"{value:+.2f}%")
+    print(printable.to_string())
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Compare each market's top10 live strategies with weekly baseline data."
+    )
+    parser.add_argument("--market", choices=["all", "us", "cn"], default="all")
+    parser.add_argument("--top", type=int, default=10)
+    parser.add_argument("--start", help="Optional start date, YYYY-MM-DD")
+    parser.add_argument("--end", help="Optional end date, YYYY-MM-DD")
+    parser.add_argument("--include-sim", action="store_true")
+    parser.add_argument("--output-dir", default="data")
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    markets = ["us", "cn"] if args.market == "all" else [args.market]
+    output_dir = Path(args.output_dir)
+
+    for market in markets:
+        values, returns, warnings = build_market_report(
+            market=market,
+            top=args.top,
+            include_sim=args.include_sim,
+            start=args.start,
+            end=args.end,
+        )
+        value_path, return_path = write_outputs(market, values, returns, output_dir)
+        print_returns(market, returns)
+        print(f"  values:  {value_path}")
+        print(f"  returns: {return_path}")
+        for warning in warnings:
+            print(f"  warning: {warning}")
+
+
+if __name__ == "__main__":
+    main()