feat: I5c MMP minute-factor study (exploratory)

config/phase_i5c_mmp_minute_factor.yaml

@@ -0,0 +1,122 @@
+# Phase I5c — MMP minute-factor study (EXPLORATORY, NOT a performance claim).
+#
+# Runs ONE user-proposed minute factor — Minute Microstructure Pressure (MMP) —
+# as a PIT-safe daily score through the I5a intraday tail event model with I5b raw
+# stk_limit execution feasibility ON. Per 1min bar t:
+#   mid=(high+low)/2; S=(close-mid)/mid; V=sqrt(volume/median(vol[t-20:t]));
+#   B=|close-open|/(high-low+eps); R=(high-low)/(mean(hl[t-20:t])+eps);
+#   MMP_t = S*V*B*R   (eps=1e-6)
+# Daily score intraday_mmp20_ew_0930_1450 = EQUAL-WEIGHT mean of valid MMP_t over
+# bars visible at the 14:50 cutoff (rolling baselines use only prior 20 bars within
+# the same symbol/day session; first 20 bars of a day are NaN). No tuning, no
+# learned/IC weights, no robustness matrix.
+#
+# 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. Same short SSE50 SH/SZ covered
+# window as I5b. The report writes to its own name/title so it never overwrites the
+# accepted I5a/I5b artifacts.
+
+project:
+  name: quantitative_trading_phase_i5c_mmp_minute_factor
+  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: i5c_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
+# mmp_ew feature (selected via intraday.score_feature), not a config factor.
+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
+  # I5c: select the exploratory MMP minute factor as the PIT-safe daily score.
+  score_feature: mmp_ew
+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_i5c_mmp_minute_factor
+  intraday_report_title: Phase I5c — MMP Minute Factor Study

data/clean/intraday_aggregate.py

@@ -47,18 +47,88 @@
 DATE_LEVEL = "date"
 DAILY_INDEX_NAMES: list[str] = [DATE_LEVEL, SYMBOL_LEVEL]
 
-# Feature keys (selectable via the ``features`` argument); column NAMES below
-# encode the cutoff and are derived from the time arguments.
+# All selectable feature keys (validated against the ``features`` argument and
+# mirrored by qt.config.IntradayCfg.score_feature). Column NAMES below encode the
+# cutoff and are derived from the time arguments.
 INTRADAY_FEATURE_KEYS: tuple[str, ...] = (
     "ret",
     "realized_vol",
     "vwap",
     "last30m_ret",
+    "mmp_ew",
+)
+# Default feature set when ``features`` is None — the original cheap four. ``mmp_ew``
+# (the I5c rolling MMP factor) is selectable-only, so the default output columns and
+# the cost of a no-args call stay EXACTLY as before (existing callers unchanged).
+DEFAULT_FEATURE_KEYS: tuple[str, ...] = (
+    "ret",
+    "realized_vol",
+    "vwap",
+    "last30m_ret",
 )
 
 DEFAULT_DECISION_TIME = "14:50:00"
 DEFAULT_SESSION_OPEN = "09:30:00"
 DEFAULT_LAST_WINDOW_MINUTES = 30
+# Minute Microstructure Pressure (MMP, I5c): rolling baseline window (prior bars
+# t-MMP_LOOKBACK..t-1) and the default denominator epsilon. EXPLORATORY factor;
+# the window is part of the factor definition, not a tuned parameter.
+MMP_LOOKBACK = 20
+DEFAULT_EPSILON = 1e-6
+
+
+def compute_minute_mmp(
+    open_: np.ndarray,
+    high: np.ndarray,
+    low: np.ndarray,
+    close: np.ndarray,
+    volume: np.ndarray,
+    *,
+    lookback: int = MMP_LOOKBACK,
+    epsilon: float = DEFAULT_EPSILON,
+) -> np.ndarray:
+    """Per-bar Minute Microstructure Pressure ``MMP_t`` for ONE symbol/day session.
+
+    Inputs are equal-length 1D arrays ORDERED by ``bar_end`` ascending and
+    belonging to a SINGLE ``(symbol, trade_date)`` session. The rolling baselines
+    use ONLY the prior ``lookback`` bars (``t-lookback..t-1``) — never bar ``t``
+    itself, never a later bar, never the prior day's tail — so the first
+    ``lookback`` bars have NaN ``MMP``.
+
+        mid_t = (high_t + low_t) / 2
+        S_t   = (close_t - mid_t) / mid_t                  (NaN if mid_t <= 0)
+        V_t   = sqrt(volume_t / median(volume[t-lookback:t]))
+                                                           (NaN if baseline <= 0 / NaN)
+        B_t   = |close_t - open_t| / (high_t - low_t + epsilon)
+        R_t   = (high_t - low_t) / (mean(hl[t-lookback:t]) + epsilon)
+                                                           (NaN if baseline is NaN)
+        MMP_t = S_t * V_t * B_t * R_t
+
+    Invalid denominators yield NaN, never ``inf``. Pure: reads no returns / no
+    future bars / no token.
+    """
+    open_ = np.asarray(open_, dtype=float)
+    high = np.asarray(high, dtype=float)
+    low = np.asarray(low, dtype=float)
+    close = np.asarray(close, dtype=float)
+    volume = np.asarray(volume, dtype=float)
+
+    hl = high - low
+    mid = (high + low) / 2.0
+
+    # Prior-`lookback` baselines: rolling over t-lookback+1..t THEN shift(1) so
+    # position t holds the statistic of bars t-lookback..t-1 (excludes bar t).
+    med_vol = pd.Series(volume).rolling(lookback).median().shift(1).to_numpy()
+    ma_hl = pd.Series(hl).rolling(lookback).mean().shift(1).to_numpy()
+
+    with np.errstate(divide="ignore", invalid="ignore"):
+        s_t = np.where(mid > 0.0, (close - mid) / mid, np.nan)
+        ratio = np.where(med_vol > 0.0, volume / med_vol, np.nan)
+        v_t = np.sqrt(ratio)
+        b_t = np.abs(close - open_) / (hl + epsilon)
+        r_t = hl / (ma_hl + epsilon)
+        mmp = s_t * v_t * b_t * r_t
+    return mmp
 
 
 def _hhmm(time_str: str) -> str:
@@ -75,7 +145,7 @@ def _hhmm_minus(time_str: str, minutes: int) -> str:
 
 def _resolve_feature_keys(features: list[str] | None) -> list[str]:
     if features is None:
-        return list(INTRADAY_FEATURE_KEYS)
+        return list(DEFAULT_FEATURE_KEYS)
     unknown = [f for f in features if f not in INTRADAY_FEATURE_KEYS]
     if unknown:
         raise ValueError(
@@ -98,6 +168,8 @@ def _column_name(
     if key == "last30m_ret":
         start = _hhmm_minus(decision_time, last_window_minutes)
         return f"intraday_last{int(last_window_minutes)}m_ret_{start}_{c}"
+    if key == "mmp_ew":
+        return f"intraday_mmp{MMP_LOOKBACK}_ew_{o}_{c}"
     raise ValueError(f"Unhandled intraday feature key: {key!r}.")
 
 
@@ -112,7 +184,12 @@ def _empty_daily(colnames: list[str]) -> pd.DataFrame:
 
 
 def _compute_group(
-    g: pd.DataFrame, decision_time: str, last_window_minutes: int
+    g: pd.DataFrame,
+    decision_time: str,
+    last_window_minutes: int,
+    keys: list[str],
+    epsilon: float,
+    session_open: str,
 ) -> dict[str, float]:
     """Per-(date, symbol) features from already PIT-filtered, bar_end-sorted bars."""
     closes = g["close"].to_numpy(dtype=float)
@@ -143,12 +220,54 @@ def _compute_group(
     else:
         last30 = float("nan")
 
-    return {
+    out = {
         "ret": ret,
         "realized_vol": realized_vol,
         "vwap": vwap,
         "last30m_ret": last30,
     }
+    # MMP is the only rolling feature; compute it ONLY when requested (I5c).
+    if "mmp_ew" in keys:
+        out["mmp_ew"] = _mmp_ew_daily(g, epsilon, trade_date, session_open)
+    return out
+
+
+def _in_session(g: pd.DataFrame, trade_date: pd.Timestamp, session_open: str) -> pd.DataFrame:
+    """Bars whose ``bar_end`` is on/after ``session_open`` (the MMP window lower bound).
+
+    The MMP daily score aggregates over ``[session_open, decision_time]`` (the upper
+    bound is the available_time cutoff already applied upstream). Restricting to the
+    in-session bars keeps PRE-session bars out of BOTH the rolling baseline and the
+    daily mean, so the first in-session bar correctly has no prior-20 baseline.
+    """
+    session_start = trade_date + pd.Timedelta(session_open)
+    return g[g["bar_end"] >= session_start]
+
+
+def _mmp_ew_daily(
+    g: pd.DataFrame, epsilon: float, trade_date: pd.Timestamp, session_open: str
+) -> float:
+    """Equal-weight mean of valid per-minute ``MMP_t`` over one PIT-filtered group.
+
+    ``g`` is one ``(date, symbol)`` session's visible bars, sorted by ``bar_end``;
+    only the in-session bars (``bar_end >= session_open``) enter, so the rolling
+    baseline starts at the session open and the first 20 in-session bars are NaN.
+    Every valid minute ``MMP_t`` gets EQUAL weight (no extra volume weighting — the
+    volume term already lives inside ``MMP_t``). No valid minute -> NaN.
+    """
+    gs = _in_session(g, trade_date, session_open)
+    if gs.empty:
+        return float("nan")
+    mmp = compute_minute_mmp(
+        gs["open"].to_numpy(dtype=float),
+        gs["high"].to_numpy(dtype=float),
+        gs["low"].to_numpy(dtype=float),
+        gs["close"].to_numpy(dtype=float),
+        gs["volume"].to_numpy(dtype=float),
+        epsilon=epsilon,
+    )
+    valid = mmp[~np.isnan(mmp)]
+    return float(np.mean(valid)) if valid.size else float("nan")
 
 
 def asof_daily_features(
@@ -158,6 +277,7 @@ def asof_daily_features(
     session_open: str = DEFAULT_SESSION_OPEN,
     last_window_minutes: int = DEFAULT_LAST_WINDOW_MINUTES,
     features: list[str] | None = None,
+    epsilon: float = DEFAULT_EPSILON,
 ) -> pd.DataFrame:
     """PIT-safe daily features from normalized 1min ``bars``.
 
@@ -189,7 +309,9 @@ def asof_daily_features(
     index_tuples: list[tuple] = []
     data: dict[str, list[float]] = {c: [] for c in colnames}
     for (date, sym), g in visible.groupby(["trade_date", SYMBOL_LEVEL], sort=True):
-        feats = _compute_group(g, decision_time, last_window_minutes)
+        feats = _compute_group(
+            g, decision_time, last_window_minutes, keys, epsilon, session_open
+        )
         index_tuples.append((date, str(sym)))
         for key, col in zip(keys, colnames):
             data[col].append(feats[key])
@@ -198,6 +320,61 @@ def asof_daily_features(
     return pd.DataFrame(data, index=index)[colnames].sort_index()
 
 
+def mmp_valid_minute_counts(
+    bars: pd.DataFrame,
+    *,
+    decision_time: str = DEFAULT_DECISION_TIME,
+    session_open: str = DEFAULT_SESSION_OPEN,
+    epsilon: float = DEFAULT_EPSILON,
+) -> pd.Series:
+    """Per-``(date, symbol)`` count of valid (non-NaN) ``MMP_t`` minutes (I5c report).
+
+    Report-only diagnostic: applies the SAME window as the daily MMP score —
+    ``available_time <= trade_date + decision_time`` (upper bound) AND
+    ``bar_end >= trade_date + session_open`` (lower bound) — then counts the
+    in-session minutes that yielded a valid ``MMP_t`` (the first ``MMP_LOOKBACK``
+    in-session bars never do). Reuses :func:`compute_minute_mmp` so there is a
+    single MMP source of truth.
+    """
+    validate_intraday_bars(bars)
+    empty = pd.Series(
+        [], dtype=int,
+        index=pd.MultiIndex.from_arrays(
+            [pd.DatetimeIndex([]), pd.Index([], dtype=object)],
+            names=DAILY_INDEX_NAMES,
+        ),
+    )
+    if len(bars) == 0:
+        return empty
+    work = bars.reset_index()
+    work["trade_date"] = work["bar_end"].dt.normalize()
+    cutoff = work["trade_date"] + pd.Timedelta(decision_time)
+    visible = work.loc[work["available_time"] <= cutoff].copy()
+    if visible.empty:
+        return empty
+    visible = visible.sort_values([SYMBOL_LEVEL, "bar_end"])
+    index_tuples: list[tuple] = []
+    counts: list[int] = []
+    for (date, sym), g in visible.groupby(["trade_date", SYMBOL_LEVEL], sort=True):
+        gs = _in_session(g, pd.Timestamp(date).normalize(), session_open)
+        if gs.empty:
+            index_tuples.append((date, str(sym)))
+            counts.append(0)
+            continue
+        mmp = compute_minute_mmp(
+            gs["open"].to_numpy(dtype=float),
+            gs["high"].to_numpy(dtype=float),
+            gs["low"].to_numpy(dtype=float),
+            gs["close"].to_numpy(dtype=float),
+            gs["volume"].to_numpy(dtype=float),
+            epsilon=epsilon,
+        )
+        index_tuples.append((date, str(sym)))
+        counts.append(int(np.count_nonzero(~np.isnan(mmp))))
+    index = pd.MultiIndex.from_tuples(index_tuples, names=DAILY_INDEX_NAMES)
+    return pd.Series(counts, index=index, dtype=int).sort_index()
+
+
 def resample_intraday_bars(bars: pd.DataFrame, freq: str) -> pd.DataFrame:
     """Derive coarser intraday bars from normalized 1min ``bars``.