From 11f4dd95dd59a026e0c4f0abbdedf2d3470b3cd7 Mon Sep 17 00:00:00 2001 From: Gahow Wang Date: Fri, 25 Sep 2026 01:16:14 +0800 Subject: [PATCH] fix(trader): make live sessions complete, final-close based and executable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Process every completed session since the last recorded one, so monitor downtime and late data are caught up; today's bar only counts after the exchange closes. The monitor's evening phase always goes through auto and catches up at startup. - Tickers without a real closing print are valued at the last price but not traded; held tickers missing from the panel keep their last recorded price instead of being valued at zero. - Use registry parameters (live momentum strategies ran top_n=len//10=49 instead of the backtested 20). - A-share orders use 100-share board lots; per-market default capital (US $10k, CN ¥70k); HK tickers are converted to USD (HKD=X, peg fallback) and exempt from the US per-share fee. - ETF strategies only run in the US book; buy_and_hold is excluded from live because its ~$20 slices never reach the minimum order size. - CN benchmark switches to 510300.SS (Yahoo's 000300.SS is truncated). - New 'replay' command rebuilds live state files from each start date. Co-Authored-By: Claude Opus 5.5 (1M context) --- trader.py | 536 +++++++++++++++++++++++++++++++++++++--------------- universe.py | 6 +- 2 files changed, 384 insertions(+), 158 deletions(-) diff --git a/trader.py b/trader.py index 0a5f944..6f57e11 100644 --- a/trader.py +++ b/trader.py @@ -31,7 +31,9 @@ State is persisted in data/trader_{market}_{strategy}.json between runs. import argparse import json import os -from datetime import datetime, timedelta +import sys +import zoneinfo +from datetime import datetime, timedelta, timezone import numpy as np import pandas as pd @@ -89,6 +91,28 @@ MARKET_FEE_TIERED = { "us": {"fee_base": 1.88, "fee_per_share": 0.009}, } +# Minimum tradable lot per market when trading whole shares. A-shares trade in +# board lots of 100. +MARKET_LOT_SIZE = {"us": 1, "cn": 100} + +# Default live capital per market, in the market's own currency. +MONITOR_CAPITAL = {"us": 10_000.0, "cn": 70_000.0} + +# Exchange close per price panel. A bar dated today is treated as a closing +# price only once the session has closed and settled for a few minutes. +MARKET_SESSION = { + "us": ("America/New_York", 16, 0), + "etfs": ("America/New_York", 16, 0), + "cn": ("Asia/Shanghai", 15, 0), +} +SESSION_SETTLE_MINUTES = 20 + +# HK-listed tickers held in a USD book are converted with this FX series +# (HKD per 1 USD). HKD is pegged within 7.75-7.85, so the band midpoint is +# used on dates the series has no quote (Yahoo's FX feed is intermittent). +HKD_FX_TICKER = "HKD=X" +HKD_PEG_MIDPOINT = 7.80 + # --------------------------------------------------------------------------- # Strategy registry # --------------------------------------------------------------------------- @@ -233,7 +257,23 @@ MIXED_STRATEGY_EXTRA_TICKERS = { "trend_rider_v6_top10": sorted(set(ETF_UNIVERSE)), } -DEFAULT_MONITOR_STRATEGIES = list(STRATEGY_REGISTRY.keys()) +# Registered strategies the live monitor does not run, with the reason. +LIVE_EXCLUDED_STRATEGIES = { + "buy_and_hold": "equal weight across ~500 names is not investable with whole " + "shares at live capital (every slice is below the minimum trade)", +} + +DEFAULT_MONITOR_STRATEGIES = [ + name for name in STRATEGY_REGISTRY if name not in LIVE_EXCLUDED_STRATEGIES +] + + +def strategy_supports_market(strategy_name: str, market: str) -> bool: + """ETF-based strategies trade USD-listed ETFs, so they only run in the US book.""" + base_name = strategy_name.removeprefix("sim_") + if base_name in ETF_STRATEGY_UNIVERSES or base_name in MIXED_STRATEGY_EXTRA_TICKERS: + return market == "us" + return True def strategy_universe(market: str, strategy_name: str) -> tuple[list[str], str]: @@ -268,6 +308,76 @@ def filter_tradable_tickers(price_data: pd.DataFrame, tickers: list[str]) -> lis return [t for t in tickers if t in price_data.columns] +def _hk_columns(columns) -> list[str]: + return [c for c in columns if str(c).endswith(".HK")] + + +def convert_quote_currency(panel: pd.DataFrame) -> pd.DataFrame: + """Express HK-listed tickers of a USD panel in USD using the HKD=X series.""" + hk = _hk_columns(panel.columns) + if not hk: + return panel + if HKD_FX_TICKER in panel.columns: + fx = panel[HKD_FX_TICKER].ffill().bfill().fillna(HKD_PEG_MIDPOINT) + else: + print(f" WARNING: {HKD_FX_TICKER} not in panel; converting HK prices at " + f"the peg midpoint {HKD_PEG_MIDPOINT}", file=sys.stderr) + fx = pd.Series(HKD_PEG_MIDPOINT, index=panel.index) + out = panel.copy() + out[hk] = out[hk].div(fx, axis=0) + return out + + +def price_request(market: str, tickers: list[str]) -> list[str]: + """Tickers to download for a USD panel, including FX needed for conversion.""" + request = sorted(set(tickers)) + if market != "cn" and _hk_columns(request): + request.append(HKD_FX_TICKER) + return request + + +def load_strategy_prices(market: str, strategy_name: str, tickers: list[str], + benchmark: str, with_open: bool = False): + """Refresh and load a strategy's prices in its portfolio currency. + + Returns ``(close, open_or_None, close_raw)``. ``close_raw`` keeps NaN + where the exchange printed no bar that day, so callers can distinguish a + stale forward-filled price from a real one. + """ + data_market = strategy_data_market(market, strategy_name) + request = price_request(market, list(tickers) + [benchmark]) + result = data_manager.update(data_market, request, with_open=with_open) + close, open_data = result if with_open else (result, None) + close_raw = data_manager.load_raw(data_market, "close").reindex( + index=close.index, columns=close.columns, + ) + close = convert_quote_currency(close) + close_raw = convert_quote_currency(close_raw) + if open_data is not None: + open_data = convert_quote_currency(open_data) + return close, open_data, close_raw + + +def completed_sessions(close_raw: pd.DataFrame, data_market: str, benchmark: str, + now: datetime | None = None) -> pd.DatetimeIndex: + """Trading days whose closing prices are final. + + A day counts when the benchmark printed a bar on it. Today's bar is + excluded until the exchange has closed and settled, because before that + it is an intraday snapshot rather than a close. + """ + index = close_raw.index + if benchmark in close_raw.columns: + index = index[close_raw[benchmark].notna().to_numpy()] + tz_name, hour, minute = MARKET_SESSION.get(data_market, MARKET_SESSION["us"]) + now_local = (now or datetime.now(timezone.utc)).astimezone(zoneinfo.ZoneInfo(tz_name)) + today = pd.Timestamp(now_local.date()) + settled = now_local.replace(hour=hour, minute=minute, second=0, microsecond=0) + timedelta( + minutes=SESSION_SETTLE_MINUTES, + ) + return index[index < today] if now_local < settled else index[index <= today] + + # --------------------------------------------------------------------------- # Persistent state # --------------------------------------------------------------------------- @@ -358,21 +468,26 @@ def generate_target_weights(strategy, open_data: pd.DataFrame, target_date) -> d def compute_trades(holdings: dict, cash: float, target_weights: dict, prices: dict, min_trade_value: float = 50.0, - integer_shares: bool = False) -> list[dict]: + integer_shares: bool = False, lot_size: int = 1, + frozen: set | None = None) -> list[dict]: """ Compute trades needed to move from current holdings to target weights. - If integer_shares=True, share deltas are rounded to whole numbers and - buys are capped so they don't exceed available cash (accounting for - commissions). Sells are processed first to free up cash. + If integer_shares=True, share deltas are rounded toward zero to whole + multiples of ``lot_size`` (e.g. 100-share A-share board lots); a position + whose target is zero is always sold in full, odd lot included. + Tickers in ``frozen`` (no fresh price today) are neither bought nor sold. Returns list of {ticker, shares_delta, direction, est_value}. """ total = portfolio_value(holdings, prices, cash) raw = [] + frozen = frozen or set() all_tickers = set(list(target_weights.keys()) + list(holdings.keys())) for ticker in sorted(all_tickers): + if ticker in frozen: + continue target_weight = target_weights.get(ticker, 0.0) price = prices.get(ticker, 0.0) if price <= 0: @@ -384,11 +499,11 @@ def compute_trades(holdings: dict, cash: float, target_weights: dict, delta = target_shares - current_shares if integer_shares: - # Round toward zero for buys (floor), away from zero for sells (floor of abs) - if delta > 0: - delta = int(delta) # floor: don't overshoot cash + if target_weight <= 0 and current_shares > 0: + delta = -current_shares # full exit may include an odd lot else: - delta = -int(abs(delta)) # floor of magnitude: don't over-sell + # Round toward zero: never overshoot cash or over-sell. + delta = int(delta / lot_size) * lot_size trade_value = abs(delta * price) if trade_value < min_trade_value: @@ -398,7 +513,7 @@ def compute_trades(holdings: dict, cash: float, target_weights: dict, raw.append({ "ticker": ticker, - "shares_delta": int(delta) if integer_shares else round(delta, 4), + "shares_delta": int(round(delta)) if integer_shares else round(delta, 4), "direction": "BUY" if delta > 0 else "SELL", "est_value": round(trade_value, 2), "price": round(price, 2), @@ -433,7 +548,8 @@ def _per_trade_commission( def execute_trades(state: dict, trades: list[dict], prices: dict, tx_cost: float = 0.001, fixed_fee: float = 0.0, fee_base: float = 0.0, fee_per_share: float = 0.0, - trade_date: str = "", integer_shares: bool = False) -> None: + trade_date: str = "", integer_shares: bool = False, + lot_size: int = 1) -> None: """Execute trades: update holdings and cash in state, append to trade_log. When integer_shares=True, sells are executed first to free up cash, @@ -451,11 +567,15 @@ def execute_trades(state: dict, trades: list[dict], prices: dict, buys = [t for t in trades if t["shares_delta"] > 0] ordered = sells + buys if integer_shares else trades + us_fee_per_share = fee_per_share for trade in ordered: ticker = trade["ticker"] delta = trade["shares_delta"] price = prices.get(ticker, trade["price"]) cost = abs(delta * price) + # The per-share schedule applies to US-listed shares only; HK lines + # (often < $1/share) pay the proportional cost and the ticket floor. + fee_per_share = 0.0 if ticker.endswith(".HK") else us_fee_per_share commission = _per_trade_commission( abs(delta), price, tx_cost, fixed_fee, fee_base, fee_per_share, ) @@ -469,6 +589,7 @@ def execute_trades(state: dict, trades: list[dict], prices: dict, if affordable_price_unit <= 0: continue affordable = int((cash - max(fixed_fee, fee_base)) / affordable_price_unit) + affordable = affordable // lot_size * lot_size if affordable < 1: continue delta = affordable @@ -567,6 +688,90 @@ def get_prices_for_date(tickers: list[str], date_idx, price_df: pd.DataFrame) -> return {t: float(row[t]) for t in tickers if t in row.index and pd.notna(row[t])} +def _with_last_known_prices(state: dict, prices: dict) -> dict: + """Fill prices of held tickers missing from the panel with their last recorded price. + + Without this a holding that drops out of the price panel would be valued + at zero and could never be sold. + """ + missing = [t for t in state["holdings"] if t not in prices] + if not missing: + return prices + out = dict(prices) + for entry in reversed(state.get("daily_log", [])): + for ticker in missing: + if ticker not in out: + price = entry.get("holdings", {}).get(ticker, {}).get("price") + if price: + out[ticker] = float(price) + if all(t in out for t in missing): + break + still = [t for t in missing if t not in out] + print(f" WARNING: no panel price for held {missing}; " + f"{'unvalued: ' + str(still) if still else 'using last recorded prices'}") + return out + + +def _strategy_fees(args, market: str) -> dict: + tier = MARKET_FEE_TIERED.get(market, {}) + return { + "tx_cost": args.tx_cost, + "fixed_fee": args.fixed_fee if args.fixed_fee > 0 else MARKET_FEES.get(market, 0.0), + "fee_base": tier.get("fee_base", 0.0), + "fee_per_share": tier.get("fee_per_share", 0.0), + } + + +def _run_session(state: dict, strategy_name: str, market: str, day: pd.Timestamp, + tickers: list[str], benchmark: str, close_data: pd.DataFrame, + open_data: pd.DataFrame, close_raw: pd.DataFrame, fees: dict, + integer_shares: bool) -> dict: + """Signal on ``day``'s open, execute at ``day``'s close, record the day. + + Tickers without a real closing print on ``day`` (forward-filled price) are + valued but not traded. + """ + day_str = str(day.date()) + lot_size = MARKET_LOT_SIZE.get(market, 1) if integer_shares else 1 + strategy = STRATEGY_REGISTRY[strategy_name]() + target_weights = generate_target_weights(strategy, open_data.loc[:day, tickers], day) + + all_held = sorted(set(target_weights) | set(state["holdings"])) + close_prices = _with_last_known_prices( + state, get_prices_for_date(all_held + [benchmark], day, close_data), + ) + raw_row = close_raw.loc[day] + frozen = {t for t in all_held if t not in raw_row.index or pd.isna(raw_row[t])} + + pre_value = portfolio_value(state["holdings"], close_prices, state["cash"]) + trades = compute_trades( + state["holdings"], state["cash"], target_weights, close_prices, + min_trade_value=max(50, pre_value * 0.001), + integer_shares=integer_shares, lot_size=lot_size, frozen=frozen, + ) + n_log = len(state["trade_log"]) + execute_trades(state, trades, close_prices, trade_date=day_str, + integer_shares=integer_shares, lot_size=lot_size, **fees) + post_value = portfolio_value(state["holdings"], close_prices, state["cash"]) + + previous = [d for d in sorted(state["daily_equity"]) if d < day_str] + prev_eq = state["daily_equity"][previous[-1]] if previous else state["initial_capital"] + state["daily_equity"][day_str] = round(post_value, 2) + record_daily_snapshot(state, day_str, close_prices, trades, prev_eq) + state["last_morning"] = day_str + state["last_evening"] = day_str + state["pending_trades"] = None + return { + "date": day_str, + "value": post_value, + "daily_return": post_value / prev_eq - 1 if prev_eq else 0.0, + "trades": len(state["trade_log"]) - n_log, + "commission": sum(t["commission"] for t in state["trade_log"][n_log:]), + "frozen": sorted(t for t in frozen if t in state["holdings"] or t in target_weights), + "positions": len(state["holdings"]), + } + + # --------------------------------------------------------------------------- # Commands # --------------------------------------------------------------------------- @@ -575,18 +780,22 @@ def cmd_morning(args): """Morning: download open prices, generate today's trade orders.""" market = args.market strategy_name = args.strategy + if not strategy_supports_market(strategy_name, market): + print(f"{strategy_name} does not run in the {market.upper()} book; skipping.") + return tickers, benchmark = strategy_universe(market, strategy_name) - data_market = strategy_data_market(market, strategy_name) - all_tickers = sorted(set(tickers + [benchmark])) # Load or init state state = load_state(market, strategy_name) if not state: - state = init_state(market, strategy_name, args.capital) - print(f"--- Initialized new portfolio: ${args.capital:,.0f} cash ---") + capital = args.capital or MONITOR_CAPITAL.get(market, 10_000.0) + state = init_state(market, strategy_name, capital) + print(f"--- Initialized new portfolio: {capital:,.0f} cash ---") # Download data (close + open) - close_data, open_data = data_manager.update(data_market, all_tickers, with_open=True) + close_data, open_data, _ = load_strategy_prices( + market, strategy_name, tickers, benchmark, with_open=True, + ) tickers = filter_tradable_tickers(close_data, tickers) today = open_data.index[-1] @@ -601,7 +810,7 @@ def cmd_morning(args): print(f"{'='*60}") # Run strategy on open prices - strategy = STRATEGY_REGISTRY[strategy_name](top_n=max(5, len(tickers) // 10)) + strategy = STRATEGY_REGISTRY[strategy_name]() target_weights = generate_target_weights(strategy, open_data[tickers], today) # Get today's open prices for trade computation @@ -613,7 +822,8 @@ def cmd_morning(args): total = portfolio_value(state["holdings"], open_prices, state["cash"]) trades = compute_trades(state["holdings"], state["cash"], target_weights, open_prices, min_trade_value=max(50, total * 0.001), - integer_shares=args.integer_shares) + integer_shares=args.integer_shares, + lot_size=MARKET_LOT_SIZE.get(market, 1)) # Store pending state["pending_trades"] = { @@ -665,8 +875,6 @@ def cmd_evening(args): market = args.market strategy_name = args.strategy tickers, benchmark = strategy_universe(market, strategy_name) - data_market = strategy_data_market(market, strategy_name) - all_tickers = sorted(set(tickers + [benchmark])) state = load_state(market, strategy_name) if not state: @@ -685,7 +893,7 @@ def cmd_evening(args): return # Get close prices - close_data = data_manager.update(data_market, all_tickers) + close_data, _, _ = load_strategy_prices(market, strategy_name, tickers, benchmark) tickers = filter_tradable_tickers(close_data, tickers) target_date = pd.Timestamp(trade_date) @@ -693,7 +901,9 @@ def cmd_evening(args): list(pending.get("target_weights", {}).keys()) + list(state["holdings"].keys()) )) - close_prices = get_prices_for_date(all_held, target_date, close_data) + close_prices = _with_last_known_prices( + state, get_prices_for_date(all_held, target_date, close_data), + ) # Recompute trades at close prices for accurate execution trades = pending.get("trades", []) @@ -708,19 +918,15 @@ def cmd_evening(args): # Recompute trades with close prices for execution target_weights = pending.get("target_weights", {}) + lot_size = MARKET_LOT_SIZE.get(market, 1) exec_trades = compute_trades( state["holdings"], state["cash"], target_weights, close_prices, min_trade_value=max(50, pre_value * 0.001), - integer_shares=args.integer_shares + integer_shares=args.integer_shares, lot_size=lot_size, ) - - fixed_fee = args.fixed_fee if args.fixed_fee > 0 else MARKET_FEES.get(args.market, 0.0) - tier = MARKET_FEE_TIERED.get(args.market, {}) - execute_trades(state, exec_trades, close_prices, - tx_cost=args.tx_cost, fixed_fee=fixed_fee, - fee_base=tier.get("fee_base", 0.0), - fee_per_share=tier.get("fee_per_share", 0.0), - trade_date=trade_date, integer_shares=args.integer_shares) + execute_trades(state, exec_trades, close_prices, trade_date=trade_date, + integer_shares=args.integer_shares, lot_size=lot_size, + **_strategy_fees(args, market)) post_value = portfolio_value(state["holdings"], close_prices, state["cash"]) state["daily_equity"][trade_date] = round(post_value, 2) @@ -772,13 +978,13 @@ def cmd_status(args): # Get latest prices tickers, benchmark = strategy_universe(market, strategy_name) - data_market = strategy_data_market(market, strategy_name) - all_tickers = sorted(set(tickers + [benchmark])) - close_data = data_manager.update(data_market, all_tickers) + close_data, _, _ = load_strategy_prices(market, strategy_name, tickers, benchmark) last_date = close_data.index[-1] all_held = list(state["holdings"].keys()) - prices = get_prices_for_date(all_held + [benchmark], last_date, close_data) + prices = _with_last_known_prices( + state, get_prices_for_date(all_held + [benchmark], last_date, close_data), + ) total = portfolio_value(state["holdings"], prices, state["cash"]) pnl = total - state["initial_capital"] @@ -1077,11 +1283,11 @@ def cmd_simulate(args): market = args.market strategy_name = args.strategy tickers, benchmark = strategy_universe(market, strategy_name) - data_market = strategy_data_market(market, strategy_name) - all_tickers = sorted(set(tickers + [benchmark])) # Load both open and close data - close_data, open_data = data_manager.update(data_market, all_tickers, with_open=True) + close_data, open_data, _ = load_strategy_prices( + market, strategy_name, tickers, benchmark, with_open=True, + ) tickers = filter_tradable_tickers(close_data, tickers) # Date range @@ -1098,7 +1304,7 @@ def cmd_simulate(args): # Pre-compute strategy weights for the full period (much faster than per-day) # Append dummy row so shift(1) inside strategy produces valid last-day signal - strategy = STRATEGY_REGISTRY[strategy_name](top_n=max(5, len(tickers) // 10)) + strategy = STRATEGY_REGISTRY[strategy_name]() open_tickers = open_data[tickers] dummy_date = open_tickers.index[-1] + pd.Timedelta(days=1) dummy_row = pd.DataFrame( @@ -1143,15 +1349,17 @@ def cmd_simulate(args): pre_value = portfolio_value(state["holdings"], close_prices, state["cash"]) # Compute and execute trades at close prices + lot_size = MARKET_LOT_SIZE.get(market, 1) trades = compute_trades( state["holdings"], state["cash"], target_weights, close_prices, min_trade_value=max(50, pre_value * 0.001), - integer_shares=args.integer_shares + integer_shares=args.integer_shares, lot_size=lot_size, ) execute_trades(state, trades, close_prices, tx_cost=args.tx_cost, fixed_fee=args.fixed_fee, - trade_date=day_str, integer_shares=args.integer_shares) + trade_date=day_str, integer_shares=args.integer_shares, + lot_size=lot_size) post_value = portfolio_value(state["holdings"], close_prices, state["cash"]) state["daily_equity"][day_str] = round(post_value, 2) @@ -1288,8 +1496,10 @@ def cmd_monitor(args): # Banner print(f"\n{'='*60}") print(f" MONITOR MODE — {len(markets)} market(s), " - f"{len(strategies)} strategies each") - print(f" Capital: ${args.capital:,.0f} | " + f"{len(strategies)} strategies") + capital_note = (f"{args.capital:,.0f}" if args.capital else + ", ".join(f"{m.upper()} {MONITOR_CAPITAL[m]:,.0f}" for m in markets)) + print(f" Capital (new portfolios): {capital_note} | " f"Integer shares: {args.integer_shares}") for mkt, sched in market_schedules.items(): fee = MARKET_FEES.get(mkt, args.fixed_fee) @@ -1336,7 +1546,7 @@ def cmd_monitor(args): import importlib import trader as _self_mod importlib.reload(_self_mod) - current = list(_self_mod.STRATEGY_REGISTRY.keys()) + current = list(_self_mod.DEFAULT_MONITOR_STRATEGIES) return current, _self_mod.STRATEGY_REGISTRY def _run_phase(market, phase, now_utc): @@ -1365,7 +1575,8 @@ def cmd_monitor(args): print(f"[monitor] {'='*55}") market_fee = MARKET_FEES.get(market, args.fixed_fee) - for strat_name in strategies: + eligible = [s for s in strategies if strategy_supports_market(s, market)] + for strat_name in eligible: sub_args = copy.copy(args) sub_args.strategy = strat_name sub_args.market = market @@ -1376,15 +1587,10 @@ def cmd_monitor(args): if phase == "morning": cmd_morning(sub_args) else: - state = load_state(market, strat_name) - today_str = now_local.strftime("%Y-%m-%d") - if (state and state.get("last_morning") == today_str - and state.get("pending_trades")): - cmd_evening(sub_args) - else: - print(f"[monitor] Morning not run for " - f"{market.upper()}:{strat_name} — using auto") - cmd_auto(sub_args) + # Evening always goes through auto: it re-derives the open + # signal, executes at the final close and catches up any + # sessions missed while the monitor was down. + cmd_auto(sub_args) print(f"[monitor] {market.upper()}:{strat_name} {phase} OK") except KeyboardInterrupt: raise @@ -1395,7 +1601,13 @@ def cmd_monitor(args): traceback.print_exc() print(f"[monitor] {market.upper()} {phase} done — " - f"{len(strategies)} strategies\n") + f"{len(eligible)} strategies\n") + + # Catch up sessions missed while the monitor was down. cmd_auto only + # processes sessions whose close is final, so this is safe at any time. + for mkt in markets: + print(f"[monitor] Startup catch-up for {mkt.upper()}") + _run_phase(mkt, "evening", datetime.now(utc)) while True: now_utc = datetime.now(utc) @@ -1458,118 +1670,112 @@ def cmd_monitor(args): print() +def _process_sessions(args, state: dict | None, start: pd.Timestamp | None = None, + label: str = "auto") -> dict | None: + """Run every completed session not yet recorded in ``state`` (catch-up safe). + + A fresh portfolio starts at ``start`` when given, otherwise at the most + recent completed session. Returns the updated state, or None if skipped. + """ + market = args.market + strategy_name = args.strategy + if not strategy_supports_market(strategy_name, market): + print(f"[{label}] {strategy_name} does not run in the {market.upper()} book; skipping.") + return None + tickers, benchmark = strategy_universe(market, strategy_name) + close_data, open_data, close_raw = load_strategy_prices( + market, strategy_name, tickers, benchmark, with_open=True, + ) + tickers = filter_tradable_tickers(close_data, tickers) + sessions = completed_sessions(close_raw, strategy_data_market(market, strategy_name), benchmark) + if len(sessions) == 0: + print(f"[{label}] No completed session in the price panel.") + return None + + if not state: + capital = args.capital or MONITOR_CAPITAL.get(market, 10_000.0) + state = init_state(market, strategy_name, capital) + first = start if start is not None else sessions[-1] + todo = sessions[sessions >= first] + print(f"[{label}] Initialized {strategy_name} ({market.upper()}) with {capital:,.0f} cash") + else: + last = state.get("last_evening") + todo = sessions[sessions > pd.Timestamp(last)] if last else sessions[-1:] + + if len(todo) == 0: + print(f"[{label}] {strategy_name}: {state.get('last_evening')} already processed.") + return state + if len(todo) > 1: + print(f"[{label}] {strategy_name}: processing {len(todo)} sessions " + f"{todo[0].date()} .. {todo[-1].date()}") + + fees = _strategy_fees(args, market) + for day in todo: + r = _run_session(state, strategy_name, market, day, tickers, benchmark, + close_data, open_data, close_raw, fees, args.integer_shares) + frozen = f" | no fresh price: {', '.join(r['frozen'])}" if r["frozen"] else "" + print(f"[{label}] {r['date']} {strategy_name:<34} value {r['value']:>12,.2f} " + f"({r['daily_return']:+.2%}) trades {r['trades']:>2} " + f"fee {r['commission']:>6.2f} positions {r['positions']:>2}{frozen}") + return state + + def cmd_auto(args): """ - Automated daily run — single invocation handles both morning + evening. + Automated daily run — signal on each session's open, execute at its close. - Designed for cron/systemd: run once after market close each trading day. - Downloads open + close prices for today, generates signals, executes, records. + Processes every completed session since the last recorded one, so missed + days (monitor downtime, late data) are caught up in order. A session is + only processed after the exchange has closed. Usage with cron (US market, run at 5pm ET weekdays): 0 17 * * 1-5 cd /path/to/quant && uv run python trader.py auto --market us For CN market (run at 4pm CST): 0 16 * * 1-5 cd /path/to/quant && uv run python trader.py auto --market cn """ - import time + state = load_state(args.market, args.strategy) + state = _process_sessions(args, state) + if state is not None: + save_state(state, args.market, args.strategy) + pnl = state["daily_equity"][state["last_evening"]] - state["initial_capital"] + print(f"[auto] P&L since inception: {pnl:>+,.2f} " + f"({pnl / state['initial_capital']:+.2%}) | state: " + f"{_state_path(args.market, args.strategy)}") - market = args.market - strategy_name = args.strategy - tickers, benchmark = strategy_universe(market, strategy_name) - data_market = strategy_data_market(market, strategy_name) - all_tickers = sorted(set(tickers + [benchmark])) - # Load or init state - state = load_state(market, strategy_name) - if not state: - state = init_state(market, strategy_name, args.capital) - print(f"[auto] Initialized new portfolio: ${args.capital:,.0f} cash") +def cmd_replay(args): + """Rebuild live state files by replaying every session from each strategy's start. - # Download data (close + open) - close_data, open_data = data_manager.update(data_market, all_tickers, with_open=True) - tickers = filter_tradable_tickers(close_data, tickers) - - today = close_data.index[-1] - today_str = str(today.date()) - - # Skip if already processed today - if state.get("last_evening") == today_str: - print(f"[auto] {today_str} already processed. Nothing to do.") - return - - print(f"\n[auto] {'='*55}") - print(f"[auto] {today_str} | {strategy_name} | {market.upper()}") - print(f"[auto] {'='*55}") - - # --- MORNING PHASE: Generate target weights from open prices --- - strategy = STRATEGY_REGISTRY[strategy_name](top_n=max(5, len(tickers) // 10)) - target_weights = generate_target_weights(strategy, open_data[tickers], today) - - all_held = list(set(list(target_weights.keys()) + list(state["holdings"].keys()))) - close_prices = get_prices_for_date(all_held + [benchmark], today, close_data) - - if not close_prices: - print(f"[auto] No close prices for {today_str}. Market likely closed.") - return - - pre_value = portfolio_value(state["holdings"], close_prices, state["cash"]) - print(f"[auto] Pre-trade value: ${pre_value:,.2f}") - print(f"[auto] Target positions: {len(target_weights)}") - - # --- EVENING PHASE: Execute trades at close prices --- - trades = compute_trades( - state["holdings"], state["cash"], target_weights, - close_prices, min_trade_value=max(50, pre_value * 0.001), - integer_shares=args.integer_shares + Uses exactly the per-session logic of ``auto``. Existing state files are + replaced; their original start date is kept unless ``--start`` is given. + """ + import copy + names = args.strategy or sorted( + os.path.basename(p)[len(f"trader_{args.market}_"):-len(".json")] + for p in __import__("glob").glob(os.path.join("data", f"trader_{args.market}_*.json")) ) - - # Fall back to per-market fee when the user didn't explicitly override - fixed_fee = args.fixed_fee if args.fixed_fee > 0 else MARKET_FEES.get(market, 0.0) - tier = MARKET_FEE_TIERED.get(market, {}) - execute_trades(state, trades, close_prices, - tx_cost=args.tx_cost, fixed_fee=fixed_fee, - fee_base=tier.get("fee_base", 0.0), - fee_per_share=tier.get("fee_per_share", 0.0), - trade_date=today_str, integer_shares=args.integer_shares) - - post_value = portfolio_value(state["holdings"], close_prices, state["cash"]) - state["daily_equity"][today_str] = round(post_value, 2) - - # Record daily snapshot - eq_vals = list(state["daily_equity"].values()) - prev_eq = eq_vals[-2] if len(eq_vals) >= 2 else state["initial_capital"] - record_daily_snapshot(state, today_str, close_prices, trades, prev_eq) - - state["last_morning"] = today_str - state["last_evening"] = today_str - state["pending_trades"] = None - save_state(state, market, strategy_name) - - # Print summary - pnl = post_value - state["initial_capital"] - pnl_pct = pnl / state["initial_capital"] * 100 - n_trades = len(trades) - commission = sum(t.get("commission", 0) for t in state["trade_log"][-n_trades:]) if n_trades else 0 - - print(f"[auto] Trades: {n_trades} | Commission: ${commission:,.2f}") - print(f"[auto] Post-trade value: ${post_value:,.2f} | Cash: ${state['cash']:,.2f}") - print(f"[auto] P&L: ${pnl:>+,.2f} ({pnl_pct:>+.2f}%)") - print(f"[auto] Holdings: {len(state['holdings'])} positions") - - if trades: - for t in trades: - p = close_prices.get(t["ticker"], t["price"]) - print(f"[auto] {t['direction']:<4} {t['ticker']:<8} {t['shares_delta']:>+10.2f} @ {p:.2f}") - - # Benchmark - bench_eq = state["daily_equity"] - dates = sorted(bench_eq.keys()) - if len(dates) >= 2: - prev_val = bench_eq[dates[-2]] - daily_ret = (post_value / prev_val - 1) * 100 - print(f"[auto] Daily return: {daily_ret:>+.2f}%") - - print(f"[auto] State saved: {_state_path(market, strategy_name)}") - print() + for name in names: + if name.startswith("sim_"): + continue + sub = copy.copy(args) + sub.strategy = name + if name not in STRATEGY_REGISTRY or name in LIVE_EXCLUDED_STRATEGIES \ + or not strategy_supports_market(name, args.market): + print(f"[replay] {args.market.upper()}:{name} is not a live strategy here; " + f"removing its state file") + if os.path.exists(_state_path(args.market, name)): + os.remove(_state_path(args.market, name)) + continue + old = load_state(args.market, name) + start = pd.Timestamp(args.start) if args.start else ( + pd.Timestamp(min(old["daily_equity"])) if old.get("daily_equity") else None + ) + if start is None: + print(f"[replay] {name}: no start date; skipping") + continue + state = _process_sessions(sub, None, start=start, label="replay") + if state is not None: + save_state(state, args.market, name) # --------------------------------------------------------------------------- @@ -1617,7 +1823,8 @@ def main(): choices=list(STRATEGY_REGISTRY.keys()), default=DEFAULT_MONITOR_STRATEGIES, help="Strategies to run (default: ALL)") - p_monitor.add_argument("--capital", type=float, default=10_000) + p_monitor.add_argument("--capital", type=float, default=None, + help="Capital for new portfolios (default: per-market MONITOR_CAPITAL)") p_monitor.add_argument("--tx-cost", type=float, default=0.001, help="Proportional transaction cost (default: 0.001 = 10bps)") p_monitor.add_argument("--fixed-fee", type=float, default=2.0, @@ -1625,6 +1832,21 @@ def main(): p_monitor.add_argument("--integer-shares", action="store_true", default=True, help="Only trade whole shares (default: True)") + # Replay: rebuild live state files from each strategy's start date + p_replay = sub.add_parser("replay", + help="Rebuild live state files by replaying all sessions (replaces state)") + p_replay.add_argument("--market", choices=list(UNIVERSES.keys()), default="us") + p_replay.add_argument("--strategy", nargs="+", default=None, + help="Strategies to replay (default: every existing state file)") + p_replay.add_argument("--start", default=None, + help="Start date (default: each state file's first recorded day)") + p_replay.add_argument("--capital", type=float, default=None, + help="Initial capital (default: per-market MONITOR_CAPITAL)") + p_replay.add_argument("--tx-cost", type=float, default=0.001) + p_replay.add_argument("--fixed-fee", type=float, default=0.0, + help="Per-trade fee floor (default: per-market MARKET_FEES)") + p_replay.add_argument("--integer-shares", action=argparse.BooleanOptionalAction, default=True) + # Status (strategy is a free-form string to allow sim_ prefixed names) p_status = sub.add_parser("status", help="Show current portfolio") p_status.add_argument("--market", choices=UNIVERSES.keys(), default="us") @@ -1666,6 +1888,8 @@ def main(): cmd_auto(args) elif args.command == "monitor": cmd_monitor(args) + elif args.command == "replay": + cmd_replay(args) elif args.command == "status": cmd_status(args) elif args.command == "simulate": diff --git a/universe.py b/universe.py index 0d6e5be..d1b88ad 100644 --- a/universe.py +++ b/universe.py @@ -129,7 +129,9 @@ UNIVERSES = { }, "cn": { "fetch": get_csi300, - "benchmark": "000300.SS", - "benchmark_label": "CSI 300 (Benchmark)", + # Yahoo's 000300.SS index series is truncated (and at times a single + # row); the CSI 300 ETF has a continuous history back to 2012. + "benchmark": "510300.SS", + "benchmark_label": "CSI 300 ETF 510300 (Benchmark)", }, }