From 90f29fbf035bf6e09ccbddbbb179367071ec96c6 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Sat, 20 Jun 2026 23:58:56 +0800 Subject: [PATCH] feat(runtime): add report-only intraday liquidity diagnostics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P-I5f: an opt-in, REPORT-ONLY execution liquidity diagnostic for the intraday tail-rebalance path. For each desired rebalance trade (|target_weight - current_weight|) it sizes the trade against the SELECTED execution-minute 1min bar's traded amount (RMB) at a max participation rate and reports a capacity ratio. It uses ONLY the execution-minute bar already chosen by the tail execution model — never a future bar, daily amount/volume/close, or any EOD proxy. Report-only: it never changes fills, can_buy/can_sell, blocked reasons, target weights, achieved holdings, turnover, cost, NAV, factor scores, MMP grouping, alpha, or portfolio construction. Default-off, so every existing config validates and behaves byte-identically. - qt/config.py: nested LiquidityDiagnosticsCfg under intraday (enabled=false default; when enabled requires positive portfolio_notional, participation in (0,1], and only mode=report_only — each with a readable error). - runtime/backtest/engine.py: additive, default-off record_rebalance_plan flag + rebalance_plan_log() capturing the (target, current) weights the engine used; default False keeps the loop byte-identical (no plan rows, no behaviour change). - runtime/backtest/event_models.py: additive accessors exposing the raw up/down limit blocked-key sets so the diagnostic skips already-blocked trades without reclassifying their original reason. - runtime/intraday_liquidity.py (new): pure formula primitives + the report-only builder (feasibility-blocked excluded with original reason kept; missing/NaN/<=0 amount reported as missing capacity data; zero desired trade skipped). - qt/intraday_tail_framework.py: wires the diagnostic into the intraday-tail report path only when enabled; new report section states it did not alter fills/NAV. - config/phase_i5f_intraday_liquidity_diagnostics.yaml (new): SSE50 smoke based on the I5b config, illustrative 10,000,000 RMB sizing. - tests/test_i5f_intraday_liquidity.py (new): formula, config validation, and report-only invariants (enabling the plan log does not change NAV/turnover/cost/ holdings; capacity uses the execution-minute amount only; existing blocks keep their original reasons). --- ...se_i5f_intraday_liquidity_diagnostics.yaml | 128 +++++++ qt/config.py | 48 +++ qt/intraday_tail_framework.py | 111 ++++++ runtime/backtest/engine.py | 49 +++ runtime/backtest/event_models.py | 14 + runtime/intraday_liquidity.py | 295 +++++++++++++++ tests/test_i5f_intraday_liquidity.py | 356 ++++++++++++++++++ 7 files changed, 1001 insertions(+) create mode 100644 config/phase_i5f_intraday_liquidity_diagnostics.yaml create mode 100644 runtime/intraday_liquidity.py create mode 100644 tests/test_i5f_intraday_liquidity.py diff --git a/config/phase_i5f_intraday_liquidity_diagnostics.yaml b/config/phase_i5f_intraday_liquidity_diagnostics.yaml new file mode 100644 index 0000000..d2bc4b3 --- /dev/null +++ b/config/phase_i5f_intraday_liquidity_diagnostics.yaml @@ -0,0 +1,128 @@ +# Phase I5f — intraday execution liquidity diagnostics (REPORT-ONLY, NOT research). +# +# Same intraday tail-rebalance smoke as I5a/I5b (decision 14:50, execute at the +# first valid 1min close in [14:51, 14:56:59], exec-to-exec holding returns, +# PIT-safe I3 score, raw stk_limit execution-time feasibility ON) PLUS an opt-in, +# REPORT-ONLY execution liquidity diagnostic. For each desired rebalance trade +# (|target_weight - current_weight|) it sizes the trade against the SELECTED +# execution-minute 1min bar's traded amount (RMB) at max_participation_rate and +# reports a capacity ratio. It NEVER changes fills, can_buy/can_sell, blocked +# reasons, target weights, achieved holdings, turnover, cost, or NAV — the backtest +# numbers are identical to running this config with the diagnostic disabled. +# +# Minute bars are read from the EXISTING intraday cache only (read-only; a miss is a +# hard blocker -> zero stk_mins live calls). Raw stk_limit flows through the same P4 +# read-through cache as the daily endpoints. SSE50 (000016.SH) over the same short, +# recent window as I5b matches the available minute-cache coverage. +# +# portfolio_notional is an ILLUSTRATIVE diagnostic sizing (10,000,000 RMB), not a +# performance or tradability claim. No new factors, no tuning. The report writes to +# its own name so it never overwrites the accepted I5a/I5b artifacts. + +project: + name: quantitative_trading_phase_i5f_intraday_liquidity_diagnostics + timezone: Asia/Shanghai +data: + source: tushare + freq: D + start: '2026-03-03' + end: '2026-06-12' + external_secret_file: /home/shaofl/Projects/financial_projects/.config.json + tushare_token_key: tushare.token + output_name: i5f_daily + cache: + enabled: true + root_dir: artifacts/cache/tushare/v1 + refresh_recent_days: 14 + refresh_dimension_days: 30 + force_refresh: [] +universe: + type: index + index_code: 000016.SH + symbols: [] + min_listing_days: 60 + filters: + # Selection-level daily tradability flags stay DISABLED (as in I5b): the daily + # close limit flag is an EOD value not known at 14:50. The honest direction-aware + # blocking is execution-time raw stk_limit vs the raw execution-minute close. + missing_close: true + suspended: false + st: false + limit_up_down: false +# factors is required by the schema but the runner ignores it: the score is the I3 +# intraday_ret feature, not a config factor. Minimal valid placeholder. +factors: +- name: momentum_20 + enabled: true + params: + window: 20 + price_col: close +processing: + drop_missing: true + standardize: + enabled: true + method: zscore + winsorize: + enabled: false + method: mad + n: 3.0 + neutralize: + enabled: false + industry_col: industry + size_col: market_cap + industry_level: L1 +alpha: + model: equal_weight + params: {} +portfolio: + constructor: topn_equal_weight + top_n: 10 + long_only: true + max_weight: null + turnover_cap: null +backtest: + initial_nav: 1.0 + rebalance: monthly + event_order: intraday_tail_rebalance + cash_return: 0.0 +intraday: + enabled: true + decision_time: '14:50:00' + data_lag: 1min + session_open: '09:30:00' + execution_model: next_minute_close + execution_window: + - '14:51:00' + - '14:56:59' + require_cache_coverage: true + missing_execution: block + # I5b execution-time raw price-limit feasibility stays ON. + price_limit_check: true + require_price_limit_coverage: true + limit_tolerance: 1.0e-06 + # I5f: opt-in, report-only execution liquidity diagnostics. + liquidity_diagnostics: + enabled: true + portfolio_notional: 10000000 + max_participation_rate: 0.05 + mode: report_only +cost: + fee_rate: 0.001 + slippage_rate: 0.0 + turnover_formula: l1 +analytics: + forward_return_periods: + - 1 + - 5 + - 20 + quantiles: 5 + benchmark: null +output: + root_dir: artifacts + data_dir: artifacts/data + factor_dir: artifacts/factors + report_dir: artifacts/reports + log_dir: artifacts/logs + overwrite: true + intraday_report_name: phase_i5f_intraday_liquidity_diagnostics + intraday_report_title: Phase I5f -- Intraday Execution Liquidity Diagnostics (report-only) diff --git a/qt/config.py b/qt/config.py index fe58360..f1b15fb 100644 --- a/qt/config.py +++ b/qt/config.py @@ -242,6 +242,48 @@ def _parse_hms(value: str) -> int: return h * 3600 + m * 60 + s +class LiquidityDiagnosticsCfg(_Strict): + """Opt-in, report-only intraday execution liquidity diagnostics (I5f). + + OFF by default, so every existing config validates and behaves unchanged. + When enabled, the intraday-tail report estimates — per desired rebalance trade + — whether the SELECTED execution-minute 1min bar's traded ``amount`` (RMB) can + absorb the trade at ``max_participation_rate``. This is REPORT-ONLY: it never + changes fills, can_buy/can_sell, blocked reasons, target weights, achieved + holdings, turnover, cost, NAV, factor scores, MMP grouping, alpha, or portfolio + construction. Only ``mode='report_only'`` is supported; an enforcement mode + fails readably (this layer must not move a real trade). + """ + + enabled: bool = False + portfolio_notional: float | None = None + max_participation_rate: float = 0.05 + mode: str = "report_only" + + @model_validator(mode="after") + def _check(self) -> "LiquidityDiagnosticsCfg": + if not self.enabled: + return self + if self.mode != "report_only": + raise ValueError( + "intraday.liquidity_diagnostics.mode only supports 'report_only' " + "(report-only diagnostics never change fills/NAV; no enforcement " + f"mode is implemented); got {self.mode!r}." + ) + if self.portfolio_notional is None or float(self.portfolio_notional) <= 0.0: + raise ValueError( + "intraday.liquidity_diagnostics.portfolio_notional must be a " + "positive RMB number when enabled (it scales the desired trade " + f"notional); got {self.portfolio_notional!r}." + ) + if not (0.0 < float(self.max_participation_rate) <= 1.0): + raise ValueError( + "intraday.liquidity_diagnostics.max_participation_rate must be in " + f"(0, 1]; got {self.max_participation_rate!r}." + ) + return self + + class IntradayCfg(_Strict): """Opt-in intraday tail-rebalance event model declaration (I5a). @@ -280,6 +322,12 @@ class IntradayCfg(_Strict): price_limit_check: bool = False require_price_limit_coverage: bool = True limit_tolerance: float = 1e-6 + # I5f: opt-in, report-only execution liquidity diagnostics. Default-off nested + # block; with the default it changes nothing (existing configs validate and + # behave byte-identically). + liquidity_diagnostics: LiquidityDiagnosticsCfg = Field( + default_factory=LiquidityDiagnosticsCfg + ) @field_validator("limit_tolerance") @classmethod diff --git a/qt/intraday_tail_framework.py b/qt/intraday_tail_framework.py index cef262c..fe2f304 100644 --- a/qt/intraday_tail_framework.py +++ b/qt/intraday_tail_framework.py @@ -50,6 +50,10 @@ from runtime.backtest.event_models import IntradayTailEventModel from runtime.backtest.sim_execution import SimExecution from runtime.intraday_execution import IntradayExecutionConfig +from runtime.intraday_liquidity import ( + LiquidityDiagnostics, + build_liquidity_diagnostics, +) _LOGGER_NAME = "qt.intraday_tail_framework" _DEFAULT_REPORT_NAME = "phase_i5a_intraday_tail_framework" @@ -140,6 +144,8 @@ class I5aResult: score_coverage: dict[str, int] # {rows, valid, nan} over the daily score panel minute_count_summary: dict[str, float] | None # valid-MMP-minutes-per-(date,symbol) distribution factor_diagnostics: tuple[dict, ...] # per settled rebalance: n / stats / Spearman IC + # I5f execution liquidity diagnostics (report-only). None unless enabled. + liquidity_diagnostics: LiquidityDiagnostics | None report_path: Path log_path: Path @@ -423,6 +429,7 @@ def run_phase_i5a_intraday(config_path: str) -> I5aResult: require_price_limit_coverage=ic.require_price_limit_coverage, ) execution = SimExecution(fee_rate=cfg.cost.fee_rate) + ld_cfg = ic.liquidity_diagnostics engine = BacktestEngine( model=model, universe=universe, @@ -432,10 +439,38 @@ def run_phase_i5a_intraday(config_path: str) -> I5aResult: selection_panel=panel, initial_nav=cfg.backtest.initial_nav, cash_return=cfg.backtest.cash_return, + # I5f: record the per-rebalance (target, current) plan ONLY when the + # report-only liquidity diagnostics are enabled. Default-off keeps the + # engine loop byte-identical (no plan rows, no behaviour change). + record_rebalance_plan=ld_cfg.enabled, ) nav_table = engine.run() logger.info("intraday backtest: %d settled periods", len(nav_table)) + # I5f execution liquidity diagnostics — REPORT-ONLY, computed AFTER settlement + # from logs the backtest already produced; it never altered fills or NAV. + liquidity_diagnostics = None + if ld_cfg.enabled: + liquidity_diagnostics = build_liquidity_diagnostics( + plan_log=engine.rebalance_plan_log(), + fills=model.fills(), + up_blocked_buy_keys=model.up_limit_blocked_buy_keys(), + down_blocked_sell_keys=model.down_limit_blocked_sell_keys(), + bars=bars, + portfolio_notional=ld_cfg.portfolio_notional, + max_participation_rate=ld_cfg.max_participation_rate, + ) + logger.info( + "intraday liquidity diagnostics (report-only): %d desired trades, " + "%d feasibility-blocked, %d missing-capacity, %d inspected, %d below 1.0x " + "capacity (did NOT alter fills/NAV)", + liquidity_diagnostics.total_desired_trades, + liquidity_diagnostics.feasibility_blocked, + liquidity_diagnostics.missing_capacity_rows, + liquidity_diagnostics.inspected, + liquidity_diagnostics.below_capacity, + ) + blocked = model.blocked_fills() blocked_counts: dict[str, int] = {} for f in blocked: @@ -490,6 +525,7 @@ def run_phase_i5a_intraday(config_path: str) -> I5aResult: score_coverage=score_coverage, minute_count_summary=minute_count_summary, factor_diagnostics=tuple(factor_diag), + liquidity_diagnostics=liquidity_diagnostics, report_path=report_path, log_path=log_path, ) @@ -598,6 +634,80 @@ def _append_factor_section(lines: list[str], result: I5aResult) -> None: lines.append("") +def _append_liquidity_section(lines: list[str], result: I5aResult) -> None: + """Report-only execution liquidity diagnostics (I5f). No-op if disabled.""" + ld = result.liquidity_diagnostics + if ld is None: + return + lines.append("## Execution liquidity diagnostics (I5f, report-only)") + lines.append("") + lines.append( + "**Report-only: these diagnostics did NOT alter fills, can_buy/can_sell, " + "blocked reasons, target weights, achieved holdings, turnover, cost, or NAV.** " + "They size each desired rebalance trade against the SELECTED execution-minute " + "1min bar's traded `amount` (RMB) at a participation cap — no future bar, no " + "daily amount/volume/close, no EOD proxy." + ) + lines.append("") + lines.append( + f"- desired trade notional = `|target_weight - current_weight| * " + f"portfolio_notional`; portfolio_notional = `{ld.portfolio_notional:,.0f}` RMB " + "(illustrative diagnostic sizing, NOT a performance claim)." + ) + lines.append( + f"- bar capacity = `execution_minute_amount * max_participation_rate`; " + f"max_participation_rate = `{ld.max_participation_rate}`." + ) + lines.append( + "- capacity_ratio = bar_capacity / desired_notional; `>= 1` ⇒ the capped bar " + "covers the trade, `< 1` ⇒ potentially liquidity-constrained." + ) + lines.append("") + lines.append(f"- total desired trades inspected: {ld.total_desired_trades}") + lines.append( + f"- excluded — existing execution feasibility already blocked them " + f"(missing bar / missing price / raw stk_limit; original reason kept, NOT " + f"reclassified as liquidity): {ld.feasibility_blocked}" + ) + lines.append(f"- missing capacity-data rows (amount missing/NaN/≤0): {ld.missing_capacity_rows}") + lines.append(f"- trades with a usable capacity ratio: {ld.inspected}") + lines.append(f"- trades below 100% capacity (ratio < 1.0): {ld.below_capacity}") + rs = ld.ratio_stats + if ld.inspected > 0: + lines.append( + f"- capacity ratio distribution — min `{rs['min']:.3f}` / p10 " + f"`{rs['p10']:.3f}` / median `{rs['median']:.3f}` / p90 `{rs['p90']:.3f}`." + ) + else: + lines.append("- capacity ratio distribution: _no inspected trade with a usable ratio._") + lines.append("") + if ld.top_constrained: + lines.append( + "Top constrained trades (lowest capacity ratio first; report-only):" + ) + lines.append("") + lines.append( + "| date | symbol | direction | desired_notional | bar_capacity_notional | capacity_ratio |" + ) + lines.append("|---|---|---|---|---|---|") + for t in ld.top_constrained: + lines.append( + f"| {pd.Timestamp(t.date).date()} | {t.symbol} | {t.direction} | " + f"{t.desired_notional:,.0f} | {t.bar_capacity_notional:,.0f} | " + f"{t.capacity_ratio:.3f} |" + ) + else: + lines.append("_no constrained trade to list._") + lines.append("") + lines.append( + "- **No alpha / performance / tradability claim** is made from this " + "diagnostic: it only flags where a single execution minute may be too thin " + "for the sized trade; partial-fill / volume-cap enforcement is explicitly " + "out of scope here." + ) + lines.append("") + + def _write_report(result: I5aResult, *, elapsed: float) -> None: """Write the I5a architecture-smoke markdown report (auditable event basis).""" cfg = result.config @@ -770,6 +880,7 @@ def _write_report(result: I5aResult, *, elapsed: float) -> None: "rule; strict mode fails before any result — never a silent passed check)" ) lines.append("") + _append_liquidity_section(lines, result) lines.append("## Achieved holdings (sample)") lines.append("") h = result.holdings_log diff --git a/runtime/backtest/engine.py b/runtime/backtest/engine.py index 8a95334..1768e9c 100644 --- a/runtime/backtest/engine.py +++ b/runtime/backtest/engine.py @@ -88,6 +88,7 @@ def __init__( selection_panel: pd.DataFrame, initial_nav: float = 1.0, cash_return: float = 0.0, + record_rebalance_plan: bool = False, ) -> None: self._model = model self._universe = universe @@ -97,9 +98,16 @@ def __init__( self._panel = selection_panel self._initial_nav = float(initial_nav) self._cash_return = float(cash_return) + # I5f (report-only): when True, record the per-rebalance (target, current) + # weights the engine actually used so the intraday-tail report can size + # desired trades for liquidity diagnostics. Purely additive bookkeeping — + # default False keeps the loop byte-identical (no plan rows recorded), so + # NAV / fills / holdings / feasibility are untouched. + self._record_rebalance_plan = bool(record_rebalance_plan) self._feasibility_log: list[dict] = [] self._holdings_log: list[dict] = [] self._event_log: list[dict] = [] + self._rebalance_plan_log: list[dict] = [] # -- run -------------------------------------------------------------- # def run(self) -> pd.DataFrame: @@ -114,6 +122,7 @@ def run(self) -> pd.DataFrame: self._feasibility_log = [] self._holdings_log = [] self._event_log = [] + self._rebalance_plan_log = [] for i, period in enumerate(periods): turnover, cost, gross, net = self._step(period) nav = nav * (1.0 + net) @@ -152,6 +161,8 @@ def _step( target = pd.Series(dtype=float) # nothing to hold -> exit what we can symbols = sorted(set(current.index) | set(target.index)) + if self._record_rebalance_plan: + self._record_rebalance_plan_rows(period.date, target, current, symbols) can_buy, can_sell = self._model.feasibility(period, symbols) self._execution.rebalance_to( target, period.date, can_buy=can_buy, can_sell=can_sell @@ -171,6 +182,44 @@ def _step( self._record_holdings(period.date, achieved) return (turnover, cost, gross, net) + # -- rebalance plan log (I5f, report-only; recorded only when requested) - # + def _record_rebalance_plan_rows( + self, + date: pd.Timestamp, + target: pd.Series, + current: pd.Series, + symbols: list[str], + ) -> None: + """Record the (target, current) weight the engine used per symbol. + + Captures EXACTLY the weights the rebalance used (``current`` is the + post-settle drifted book, ``target`` the constructor output), so a desired + trade ``target - current`` is faithful and cannot be re-derived wrongly + downstream. Report-only bookkeeping; does not affect the rebalance. + """ + for sym in symbols: + s = str(sym) + self._rebalance_plan_log.append( + { + "date": date, + "symbol": s, + "target_weight": float(target.get(s, 0.0)), + "current_weight": float(current.get(s, 0.0)), + } + ) + + def rebalance_plan_log(self) -> pd.DataFrame: + """Per-rebalance (date, symbol, target_weight, current_weight) plan. + + Empty (no rows) unless the engine was built with + ``record_rebalance_plan=True``. Report-only: never consulted by the + rebalance/settlement loop. + """ + cols = ["date", "symbol", "target_weight", "current_weight"] + if not self._rebalance_plan_log: + return pd.DataFrame(columns=cols) + return pd.DataFrame(self._rebalance_plan_log)[cols].reset_index(drop=True) + # -- feasibility log -------------------------------------------------- # def _record_feasibility(self, date: pd.Timestamp, achieved: pd.Series) -> None: fill = getattr(self._execution, "last_fill", None) diff --git a/runtime/backtest/event_models.py b/runtime/backtest/event_models.py index e93b211..a381e59 100644 --- a/runtime/backtest/event_models.py +++ b/runtime/backtest/event_models.py @@ -383,3 +383,17 @@ def down_limit_blocked_sells(self) -> int: def missing_limit_rows(self) -> int: """Count of evaluated (date, symbol) pairs with no usable raw limit row.""" return len(self._unchecked_limits) + + def up_limit_blocked_buy_keys(self) -> set[tuple[pd.Timestamp, str]]: + """(rebalance date, symbol) keys whose BUY was blocked by a raw up-limit. + + Read-only view of the idempotent diagnostic state populated during + ``feasibility()``. Exposed so a report-only liquidity diagnostic (I5f) can + skip trades the existing feasibility already blocked, without re-deriving + or reclassifying the block reason. + """ + return set(self._up_blocked_buys.keys()) + + def down_limit_blocked_sell_keys(self) -> set[tuple[pd.Timestamp, str]]: + """(rebalance date, symbol) keys whose SELL was blocked by a raw down-limit.""" + return set(self._down_blocked_sells.keys()) diff --git a/runtime/intraday_liquidity.py b/runtime/intraday_liquidity.py new file mode 100644 index 0000000..e4f17bc --- /dev/null +++ b/runtime/intraday_liquidity.py @@ -0,0 +1,295 @@ +"""Report-only intraday execution liquidity diagnostics (I5f). + +For each desired rebalance trade at a 14:50 tail rebalance — the change +``target_weight - current_weight`` the engine actually planned — estimate whether +the SELECTED execution-minute 1min bar's traded ``amount`` (RMB) can absorb the +trade at a max participation rate. This is a DIAGNOSTIC ONLY: it never changes +fills, ``can_buy``/``can_sell``, blocked reasons, target weights, achieved +holdings, turnover, cost, NAV, factor scores, MMP grouping, alpha, or portfolio +construction. The intraday-tail runner builds it AFTER the backtest has settled, +from logs the backtest already produced. + +Capacity model — execution-minute only (no future bar, no daily ``amount`` / +``volume`` / ``close``, no EOD proxy): + + desired_notional = |target_weight - current_weight| * portfolio_notional + bar_capacity_notional = execution_minute_amount * max_participation_rate + capacity_ratio = bar_capacity_notional / desired_notional + +``capacity_ratio >= 1`` means the bar (at the participation cap) covers the desired +trade; ``< 1`` flags a potentially liquidity-constrained trade. Rules: + + * a zero desired trade (``|delta| == 0``) is not a trade and is skipped (no + division by zero); + * a trade whose direction was already blocked by the existing I5a/I5b execution + feasibility (missing bar, missing price, raw ``stk_limit`` up/down limit) is + counted as EXCLUDED and keeps its ORIGINAL block reason — it is NEVER + reclassified as a liquidity block, and never gets a capacity ratio; + * a missing / NaN / non-positive execution-minute ``amount`` on an otherwise + executable trade is reported as MISSING capacity data, never inferred. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +import numpy as np +import pandas as pd + +from runtime.intraday_execution import ExecutionFill + +_PLAN_COLUMNS = ["date", "symbol", "target_weight", "current_weight"] + + +# --------------------------------------------------------------------------- # +# Small, separable, unit-testable primitives +# --------------------------------------------------------------------------- # +def trade_direction(target_weight: float, current_weight: float) -> str | None: + """``"buy"`` if the weight rises, ``"sell"`` if it falls, ``None`` if flat.""" + delta = float(target_weight) - float(current_weight) + if delta > 0.0: + return "buy" + if delta < 0.0: + return "sell" + return None + + +def desired_trade_notional( + target_weight: float, current_weight: float, portfolio_notional: float +) -> float: + """``|target - current| * portfolio_notional`` — the RMB to trade this rebalance.""" + return abs(float(target_weight) - float(current_weight)) * float(portfolio_notional) + + +def bar_capacity_notional( + amount: float | None, max_participation_rate: float +) -> float | None: + """Execution-minute RMB capacity at the participation cap. + + Returns ``None`` when ``amount`` is missing / NaN / non-positive — that is + MISSING capacity data, not zero capacity, and must never be inferred. + """ + if amount is None: + return None + a = float(amount) + if not np.isfinite(a) or a <= 0.0: + return None + return a * float(max_participation_rate) + + +def capacity_ratio(desired_notional: float, capacity_notional: float) -> float: + """``capacity_notional / desired_notional`` (caller guarantees desired > 0).""" + d = float(desired_notional) + if d <= 0.0: + raise ValueError( + "capacity_ratio requires a positive desired_notional (a zero desired " + "trade is not a trade and must be skipped upstream)." + ) + return float(capacity_notional) / d + + +# --------------------------------------------------------------------------- # +# Result containers (immutable) +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class LiquidityTrade: + """One desired rebalance trade's liquidity diagnostic (immutable, report-only).""" + + date: pd.Timestamp + symbol: str + direction: str # "buy" | "sell" + desired_notional: float + bar_capacity_notional: float | None # None when capacity data is missing/blocked + capacity_ratio: float | None # None when blocked or capacity data missing + feasibility_blocked: bool + block_reason: str | None # original execution block reason (kept as-is) + missing_capacity_data: bool + + +@dataclass(frozen=True) +class LiquidityDiagnostics: + """Aggregate report-only liquidity diagnostics over all desired trades.""" + + portfolio_notional: float + max_participation_rate: float + total_desired_trades: int + feasibility_blocked: int + missing_capacity_rows: int + inspected: int # executable trades with a usable capacity ratio + below_capacity: int # inspected trades with capacity_ratio < 1 + ratio_stats: dict[str, float] # min / p10 / median / p90 over inspected + top_constrained: tuple[LiquidityTrade, ...] + trades: tuple[LiquidityTrade, ...] + + +# --------------------------------------------------------------------------- # +# Execution-minute amount lookup (selected bar only — never a future/daily value) +# --------------------------------------------------------------------------- # +def _exec_amounts( + bars: pd.DataFrame, fills: list[ExecutionFill] +) -> dict[tuple[pd.Timestamp, str], float]: + """``{(norm date, symbol): execution-minute amount}`` for non-blocked fills. + + The amount is read at the EXACT selected execution bar (``bar_end == + exec_time``), so it is the execution-minute traded value — never a future bar, + never a daily total. Absent rows resolve to NaN and are handled as missing data + by the caller. + """ + keys: list[tuple[pd.Timestamp, str]] = [] + bar_index: list[tuple[pd.Timestamp, str]] = [] + for f in fills: + if f.blocked or f.exec_time is None: + continue + keys.append((pd.Timestamp(f.date).normalize(), str(f.symbol))) + bar_index.append((pd.Timestamp(f.exec_time), str(f.symbol))) + if not keys: + return {} + amounts = bars["amount"].reindex( + pd.MultiIndex.from_tuples(bar_index, names=list(bars.index.names)) + ).to_numpy() + return {k: amounts[i] for i, k in enumerate(keys)} + + +def _block_status( + date: pd.Timestamp, + symbol: str, + direction: str, + fill: ExecutionFill | None, + up_blocked_buy_keys: set, + down_blocked_sell_keys: set, +) -> tuple[bool, str | None]: + """(feasibility_blocked, original_reason) for one desired trade direction. + + A missing/NaN execution bar (blocked fill) blocks BOTH directions and keeps its + fill reason; otherwise a buy is blocked iff its (date, symbol) is in the raw + up-limit set, a sell iff it is in the raw down-limit set. The reason is the + ORIGINAL execution reason — never reclassified to a liquidity block. + """ + if fill is not None and fill.blocked: + return True, fill.reason + key = (date, symbol) + if direction == "buy" and key in up_blocked_buy_keys: + return True, "up_limit" + if direction == "sell" and key in down_blocked_sell_keys: + return True, "down_limit" + return False, None + + +# --------------------------------------------------------------------------- # +# Builder +# --------------------------------------------------------------------------- # +def build_liquidity_diagnostics( + *, + plan_log: pd.DataFrame, + fills: list[ExecutionFill], + up_blocked_buy_keys: set, + down_blocked_sell_keys: set, + bars: pd.DataFrame, + portfolio_notional: float, + max_participation_rate: float, + top_n: int = 10, +) -> LiquidityDiagnostics: + """Build the report-only liquidity diagnostics from settled backtest logs. + + ``plan_log`` is the engine's ``rebalance_plan_log()`` (date, symbol, + target_weight, current_weight). For each desired trade (``|delta| > 0``): if the + existing feasibility blocked that direction, count it as excluded (original + reason kept); else read the selected execution-minute ``amount`` and compute the + capacity ratio, flagging missing amount data. NEVER mutates any input or the + backtest. + """ + notional = float(portfolio_notional) + part = float(max_participation_rate) + amount_lookup = _exec_amounts(bars, fills) + fill_map = { + (pd.Timestamp(f.date).normalize(), str(f.symbol)): f for f in fills + } + + trades: list[LiquidityTrade] = [] + feasibility_blocked = 0 + missing_capacity = 0 + inspected_ratios: list[float] = [] + + if not plan_log.empty: + for _, row in plan_log.iterrows(): + date = pd.Timestamp(row["date"]).normalize() + symbol = str(row["symbol"]) + direction = trade_direction(row["target_weight"], row["current_weight"]) + if direction is None: + continue # zero desired trade -> not a trade -> skip (no div by zero) + desired = desired_trade_notional( + row["target_weight"], row["current_weight"], notional + ) + fill = fill_map.get((date, symbol)) + blocked, reason = _block_status( + date, symbol, direction, fill, + up_blocked_buy_keys, down_blocked_sell_keys, + ) + if blocked: + feasibility_blocked += 1 + trades.append( + LiquidityTrade( + date=date, symbol=symbol, direction=direction, + desired_notional=desired, bar_capacity_notional=None, + capacity_ratio=None, feasibility_blocked=True, + block_reason=reason, missing_capacity_data=False, + ) + ) + continue + amount = amount_lookup.get((date, symbol)) + cap = bar_capacity_notional(amount, part) + if cap is None: + missing_capacity += 1 + trades.append( + LiquidityTrade( + date=date, symbol=symbol, direction=direction, + desired_notional=desired, bar_capacity_notional=None, + capacity_ratio=None, feasibility_blocked=False, + block_reason=None, missing_capacity_data=True, + ) + ) + continue + ratio = capacity_ratio(desired, cap) + inspected_ratios.append(ratio) + trades.append( + LiquidityTrade( + date=date, symbol=symbol, direction=direction, + desired_notional=desired, bar_capacity_notional=cap, + capacity_ratio=ratio, feasibility_blocked=False, + block_reason=None, missing_capacity_data=False, + ) + ) + + below = sum(1 for r in inspected_ratios if r < 1.0) + ratio_stats = _ratio_stats(inspected_ratios) + inspected_trades = [t for t in trades if t.capacity_ratio is not None] + top_constrained = sorted( + inspected_trades, + key=lambda t: (t.capacity_ratio, pd.Timestamp(t.date), t.symbol), + )[: max(0, int(top_n))] + + return LiquidityDiagnostics( + portfolio_notional=notional, + max_participation_rate=part, + total_desired_trades=len(trades), + feasibility_blocked=feasibility_blocked, + missing_capacity_rows=missing_capacity, + inspected=len(inspected_ratios), + below_capacity=below, + ratio_stats=ratio_stats, + top_constrained=tuple(top_constrained), + trades=tuple(trades), + ) + + +def _ratio_stats(ratios: list[float]) -> dict[str, float]: + """min / p10 / median / p90 over the inspected capacity ratios (NaN if empty).""" + if not ratios: + return {k: float("nan") for k in ("min", "p10", "median", "p90")} + arr = np.asarray(ratios, dtype=float) + return { + "min": float(np.min(arr)), + "p10": float(np.percentile(arr, 10)), + "median": float(np.percentile(arr, 50)), + "p90": float(np.percentile(arr, 90)), + } diff --git a/tests/test_i5f_intraday_liquidity.py b/tests/test_i5f_intraday_liquidity.py new file mode 100644 index 0000000..47603e2 --- /dev/null +++ b/tests/test_i5f_intraday_liquidity.py @@ -0,0 +1,356 @@ +"""I5f: report-only intraday execution liquidity diagnostics. + +Covers the formula primitives, the config schema (default-off + readable enabled +validation), and the report-only invariants (enabling the diagnostic does not move +NAV/turnover/cost/holdings; capacity uses the execution-minute amount ONLY; existing +execution blocks keep their original reasons and are never reclassified). All +network-free: synthetic bars / fills / panels. +""" + +from __future__ import annotations + +from pathlib import Path + +import pandas as pd +import pytest +from pandas.testing import assert_frame_equal + +from data.clean.intraday_schema import normalize_intraday_bars +from data.clean.schema import normalize_panel +from qt.config import LiquidityDiagnosticsCfg, load_config +from qt.pipeline import _FrameScores +from portfolio.construct import TopNEqualWeight +from runtime.backtest.engine import BacktestEngine +from runtime.backtest.event_models import IntradayTailEventModel +from runtime.backtest.sim_execution import SimExecution +from runtime.intraday_execution import ( + REASON_MISSING_PRICE, + REASON_NO_BAR, + ExecutionFill, + IntradayExecutionConfig, +) +from runtime.intraday_liquidity import ( + bar_capacity_notional, + build_liquidity_diagnostics, + capacity_ratio, + desired_trade_notional, + trade_direction, +) +from universe.static import StaticUniverse + +_CONFIG_DIR = Path(__file__).resolve().parent.parent / "config" + +_DATES = [ + "2024-01-30", "2024-01-31", + "2024-02-28", "2024-02-29", + "2024-03-28", "2024-03-29", +] +_JAN, _FEB, _MAR = ( + pd.Timestamp("2024-01-31"), pd.Timestamp("2024-02-29"), pd.Timestamp("2024-03-29"), +) + + +# --------------------------------------------------------------------------- # +# builders +# --------------------------------------------------------------------------- # +def _daily_panel(closes: dict) -> pd.DataFrame: + rows = [ + { + "date": pd.Timestamp(d), "symbol": s, + "open": c, "high": c, "low": c, "close": c, + "volume": 1.0, "amount": 1.0, "adj_factor": 1.0, + } + for (d, s), c in closes.items() + ] + return normalize_panel(pd.DataFrame(rows)) + + +def _grid_panel(symbols: list[str]) -> pd.DataFrame: + return _daily_panel({(d, s): 100.0 for d in _DATES for s in symbols}) + + +def _minute_bars(specs: list[tuple[str, str, float, float]]) -> pd.DataFrame: + """1min bars from ``(symbol, 'YYYY-MM-DD HH:MM:SS', close, amount)`` specs.""" + rows = [ + { + "time": pd.Timestamp(t), "symbol": s, + "open": c, "high": c, "low": c, "close": c, + "volume": 1.0, "amount": float(a), "source_trade_time": t, + } + for (s, t, c, a) in specs + ] + return normalize_intraday_bars(pd.DataFrame(rows), freq="1min", data_lag="1min") + + +def _scores(map_: dict) -> _FrameScores: + idx = pd.MultiIndex.from_tuples( + [(pd.Timestamp(d), s) for (d, s) in map_], names=["date", "symbol"] + ) + return _FrameScores(pd.Series(list(map_.values()), index=idx, name="score")) + + +def _plan(rows: list[tuple]) -> pd.DataFrame: + """plan_log frame from ``(date, symbol, target_weight, current_weight)`` tuples.""" + return pd.DataFrame( + [ + {"date": pd.Timestamp(d), "symbol": s, + "target_weight": tw, "current_weight": cw} + for (d, s, tw, cw) in rows + ] + ) + + +# --------------------------------------------------------------------------- # +# 1. formula primitives +# --------------------------------------------------------------------------- # +def test_capacity_ratio_formula(): + assert desired_trade_notional(0.3, 0.1, 1_000_000) == pytest.approx(200_000.0) + assert bar_capacity_notional(1_000_000, 0.05) == pytest.approx(50_000.0) + assert capacity_ratio(200_000.0, 50_000.0) == pytest.approx(0.25) + # ratio >= 1 means the capped bar covers the trade. + assert capacity_ratio(40_000.0, 50_000.0) == pytest.approx(1.25) + + +def test_trade_direction_labeling(): + assert trade_direction(0.3, 0.1) == "buy" + assert trade_direction(0.1, 0.3) == "sell" + assert trade_direction(0.2, 0.2) is None # flat -> not a trade + + +def test_zero_desired_trade_avoids_division_by_zero(): + # capacity_ratio refuses a zero desired notional (would divide by zero). + with pytest.raises(ValueError): + capacity_ratio(0.0, 50_000.0) + # The builder simply SKIPS flat rows: only the real trade is inspected. + bars = _minute_bars([("A", f"{_JAN.date()} 14:51:00", 10.0, 1_000_000.0)]) + plan = _plan([ + (_JAN, "A", 1.0, 0.0), # real buy + (_JAN, "B", 0.5, 0.5), # flat -> skipped + ]) + fills = [ExecutionFill("A", _JAN, pd.Timestamp(f"{_JAN.date()} 14:51:00"), 10.0, False, None)] + diag = build_liquidity_diagnostics( + plan_log=plan, fills=fills, up_blocked_buy_keys=set(), + down_blocked_sell_keys=set(), bars=bars, + portfolio_notional=1_000_000, max_participation_rate=0.05, + ) + assert diag.total_desired_trades == 1 # the flat row never becomes a trade + assert diag.inspected == 1 + + +@pytest.mark.parametrize("amount", [None, float("nan"), 0.0, -5.0]) +def test_missing_nan_zero_negative_amount_is_missing_capacity(amount): + # primitive: bad amount -> None capacity (never inferred as zero). + assert bar_capacity_notional(amount, 0.05) is None + # builder: an executable trade whose exec-minute amount is bad is reported as + # MISSING capacity data, not below-capacity and not a feasibility block. None + # models "no bar row found at the exec time"; nan/0/-5 model a present-but-bad + # amount on the selected exec bar. + if amount is None: + bars = _minute_bars([("Z", f"{_JAN.date()} 14:51:00", 10.0, 1.0)]) # unrelated bar only + else: + bars = _minute_bars([("A", f"{_JAN.date()} 14:51:00", 10.0, amount)]) + plan = _plan([(_JAN, "A", 1.0, 0.0)]) + fills = [ExecutionFill("A", _JAN, pd.Timestamp(f"{_JAN.date()} 14:51:00"), 10.0, False, None)] + diag = build_liquidity_diagnostics( + plan_log=plan, fills=fills, up_blocked_buy_keys=set(), + down_blocked_sell_keys=set(), bars=bars, + portfolio_notional=1_000_000, max_participation_rate=0.05, + ) + assert diag.missing_capacity_rows == 1 + assert diag.inspected == 0 + assert diag.below_capacity == 0 + assert diag.feasibility_blocked == 0 + assert diag.trades[0].missing_capacity_data is True + assert diag.trades[0].capacity_ratio is None + + +# --------------------------------------------------------------------------- # +# 2. config schema +# --------------------------------------------------------------------------- # +def test_default_off_and_old_configs_validate_unchanged(): + # default: the nested block is present but disabled -> behaviour unchanged. + from qt.config import IntradayCfg + + ic = IntradayCfg() + assert ic.liquidity_diagnostics.enabled is False + # every shipped config still validates. + for path in sorted(_CONFIG_DIR.glob("*.yaml")): + load_config(str(path)) # raises on failure + + +def test_i5f_config_loads_and_enables(): + cfg = load_config(str(_CONFIG_DIR / "phase_i5f_intraday_liquidity_diagnostics.yaml")) + ld = cfg.intraday.liquidity_diagnostics + assert ld.enabled is True + assert ld.portfolio_notional == 10_000_000 + assert ld.max_participation_rate == 0.05 + assert ld.mode == "report_only" + + +def test_enabled_requires_positive_notional(): + with pytest.raises(ValueError, match="portfolio_notional"): + LiquidityDiagnosticsCfg(enabled=True, portfolio_notional=None) + with pytest.raises(ValueError, match="portfolio_notional"): + LiquidityDiagnosticsCfg(enabled=True, portfolio_notional=-1.0) + # positive notional is accepted. + ok = LiquidityDiagnosticsCfg(enabled=True, portfolio_notional=1_000_000) + assert ok.portfolio_notional == 1_000_000 + + +def test_invalid_participation_rate_fails_readably(): + with pytest.raises(ValueError, match="max_participation_rate"): + LiquidityDiagnosticsCfg( + enabled=True, portfolio_notional=1_000_000, max_participation_rate=0.0 + ) + with pytest.raises(ValueError, match="max_participation_rate"): + LiquidityDiagnosticsCfg( + enabled=True, portfolio_notional=1_000_000, max_participation_rate=1.5 + ) + + +def test_unsupported_mode_fails_readably(): + with pytest.raises(ValueError, match="report_only"): + LiquidityDiagnosticsCfg( + enabled=True, portfolio_notional=1_000_000, mode="enforce" + ) + + +def test_disabled_ignores_other_fields(): + # disabled -> the strict checks do not fire (defaults preserve all configs). + cfg = LiquidityDiagnosticsCfg(enabled=False, portfolio_notional=None, mode="enforce") + assert cfg.enabled is False + + +# --------------------------------------------------------------------------- # +# 3. report-only invariants (engine + helper) +# --------------------------------------------------------------------------- # +def _intraday_inputs(): + panel = _grid_panel(["A", "B"]) + universe = StaticUniverse(["A", "B"]) + scores = _scores({(d, s): {"A": 2.0, "B": 1.0}[s] for d in _DATES for s in "AB"}) + bars = _minute_bars( + [(s, f"{day.date()} 14:51:00", 10.0 + i, 1_000_000.0) + for i, day in enumerate((_JAN, _FEB, _MAR)) for s in ("A", "B")] + ) + return panel, universe, scores, bars + + +def _engine(panel, universe, scores, bars, *, record_plan): + return BacktestEngine( + model=IntradayTailEventModel( + calendar_panel=panel, bars=bars, cfg=IntradayExecutionConfig() + ), + universe=universe, + scores=scores, + constructor=TopNEqualWeight(2, long_only=True), + execution=SimExecution(fee_rate=0.001), + selection_panel=panel, + initial_nav=1.0, + cash_return=0.0, + record_rebalance_plan=record_plan, + ) + + +def test_enabling_plan_does_not_change_backtest(): + panel, universe, scores, bars = _intraday_inputs() + + base = _engine(panel, universe, scores, bars, record_plan=False) + base_nav = base.run() + + withplan = _engine(panel, universe, scores, bars, record_plan=True) + plan_nav = withplan.run() + + # NAV / turnover / cost AND holdings/feasibility are byte-identical: the plan + # log is pure bookkeeping; the report-only diagnostic never moves the book. + assert_frame_equal(base_nav, plan_nav) + assert_frame_equal(base.feasibility_log(), withplan.feasibility_log()) + assert_frame_equal(base.holdings_log(), withplan.holdings_log()) + # disabled (record_plan=False) -> NO plan rows recorded -> no diagnostic input. + assert base.rebalance_plan_log().empty + assert not withplan.rebalance_plan_log().empty + + +def test_diagnostics_use_execution_minute_amount_only(): + # The exec bar (14:51) carries amount 50e6; a LATER same-day minute carries a + # huge 999e6 that must NEVER be used. Capacity must reflect the 14:51 amount. + exec_time = pd.Timestamp(f"{_JAN.date()} 14:51:00") + bars = _minute_bars([ + ("A", f"{_JAN.date()} 14:51:00", 10.0, 50_000_000.0), # selected exec bar + ("A", f"{_JAN.date()} 14:55:00", 10.0, 999_000_000.0), # later minute: ignored + ]) + plan = _plan([(_JAN, "A", 1.0, 0.0)]) # desired = 1.0 * 1e6 = 1e6 + fills = [ExecutionFill("A", _JAN, exec_time, 10.0, False, None)] + diag = build_liquidity_diagnostics( + plan_log=plan, fills=fills, up_blocked_buy_keys=set(), + down_blocked_sell_keys=set(), bars=bars, + portfolio_notional=1_000_000, max_participation_rate=0.05, + ) + t = diag.trades[0] + # 50e6 * 0.05 = 2.5e6 (NOT 999e6 * 0.05); ratio = 2.5e6 / 1e6 = 2.5 + assert t.bar_capacity_notional == pytest.approx(2_500_000.0) + assert t.capacity_ratio == pytest.approx(2.5) + assert diag.below_capacity == 0 + + +def test_existing_blocks_keep_reason_not_reclassified(): + # U: buy blocked by raw up-limit; Dn: sell blocked by raw down-limit; + # N: missing execution bar; M: missing price. None should be reclassified as a + # liquidity (missing-capacity / below-capacity) row. + bars = _minute_bars([ + ("U", f"{_JAN.date()} 14:51:00", 10.0, 1_000_000.0), + ("Dn", f"{_JAN.date()} 14:51:00", 10.0, 1_000_000.0), + ]) + plan = _plan([ + (_JAN, "U", 1.0, 0.0), # buy + (_JAN, "Dn", 0.0, 1.0), # sell + (_JAN, "N", 1.0, 0.0), # buy, no bar + (_JAN, "M", 1.0, 0.0), # buy, missing price + ]) + fills = [ + ExecutionFill("U", _JAN, pd.Timestamp(f"{_JAN.date()} 14:51:00"), 10.0, False, None), + ExecutionFill("Dn", _JAN, pd.Timestamp(f"{_JAN.date()} 14:51:00"), 10.0, False, None), + ExecutionFill("N", _JAN, None, None, True, REASON_NO_BAR), + ExecutionFill("M", _JAN, pd.Timestamp(f"{_JAN.date()} 14:51:00"), None, True, REASON_MISSING_PRICE), + ] + diag = build_liquidity_diagnostics( + plan_log=plan, fills=fills, + up_blocked_buy_keys={(_JAN.normalize(), "U")}, + down_blocked_sell_keys={(_JAN.normalize(), "Dn")}, + bars=bars, portfolio_notional=1_000_000, max_participation_rate=0.05, + ) + by_sym = {t.symbol: t for t in diag.trades} + assert by_sym["U"].feasibility_blocked and by_sym["U"].block_reason == "up_limit" + assert by_sym["Dn"].feasibility_blocked and by_sym["Dn"].block_reason == "down_limit" + assert by_sym["N"].feasibility_blocked and by_sym["N"].block_reason == REASON_NO_BAR + assert by_sym["M"].feasibility_blocked and by_sym["M"].block_reason == REASON_MISSING_PRICE + # none reclassified as liquidity issues, none given a capacity ratio. + assert diag.feasibility_blocked == 4 + assert diag.missing_capacity_rows == 0 + assert diag.below_capacity == 0 + assert diag.inspected == 0 + assert all(t.capacity_ratio is None for t in diag.trades) + + +def test_below_capacity_flagged_and_top_constrained_sorted(): + exec_time = pd.Timestamp(f"{_JAN.date()} 14:51:00") + # Thin bar (amount 5e6 -> capacity 250k) vs a large desired trade (1e6) -> ratio 0.25. + bars = _minute_bars([ + ("A", f"{_JAN.date()} 14:51:00", 10.0, 5_000_000.0), # ratio 0.25 (constrained) + ("B", f"{_JAN.date()} 14:51:00", 10.0, 100_000_000.0), # ratio 5.0 (ample) + ]) + plan = _plan([(_JAN, "A", 1.0, 0.0), (_JAN, "B", 1.0, 0.0)]) + fills = [ + ExecutionFill("A", _JAN, exec_time, 10.0, False, None), + ExecutionFill("B", _JAN, exec_time, 10.0, False, None), + ] + diag = build_liquidity_diagnostics( + plan_log=plan, fills=fills, up_blocked_buy_keys=set(), + down_blocked_sell_keys=set(), bars=bars, + portfolio_notional=1_000_000, max_participation_rate=0.05, + ) + assert diag.inspected == 2 + assert diag.below_capacity == 1 # only A < 1.0 + assert diag.ratio_stats["min"] == pytest.approx(0.25) + # top constrained lists the lowest ratio first. + assert diag.top_constrained[0].symbol == "A" + assert diag.top_constrained[0].capacity_ratio == pytest.approx(0.25)