refactor(strategy): 定寸口径全量统一_ex总资产语义(用户08-26拍板「都要统一ex的」)——回测侧四原版per_value现金→_total_value(momentum/value/small_cap同款total<=0守卫+all_weather两处:补跌加仓n_pick/月度铺余位空槽数),all_weather_ex同步(08-25 A修法时被排除的两处补齐,移交文档「_ex已全部用」与实际不符);value_selection_ex交集定序list→sorted对齐fa04475(原版已sorted,_ex副本漏带=原版↔_ex同参一致的最后一处差);8策略类inspect验证零残留现金口径;测试参数化原版+_ex双跑(12绿=统一验收)+aw两现场新增;顺修test_live_reconcile写死日期跨午夜翻红(08-26实锤,改动态今天);524全绿;回测历史结果将变(目的=回测↔实盘对齐) [vps]

sanguo_portfolio/strategies/all_weather.py

@@ -201,7 +201,10 @@ class AllWeatherStrategy:
             # 跌幅最负的 n_pick 只(即"补跌最多")
             idx_sorted = np.argsort(drops)[:n_pick]
             picked = [remaining[i] for i in idx_sorted]
-            cash = _available_cash(context) / n_pick
+            total = _total_value(context)
+            if total <= 0:
+                return
+            cash = total / n_pick
             for code in picked:
                 self.broker.order_value(code, cash)
                 logger.debug("补跌最多的N支 Order %s", code)
@@ -270,7 +273,10 @@ class AllWeatherStrategy:
         position_count = len(positions)
         target_num = len(target)
         if target_num > position_count:
-            cash = _available_cash(context) / (target_num - position_count)
+            total = _total_value(context)
+            if total <= 0:
+                return
+            cash = total / (target_num - position_count)
             for stock in target:
                 if stock in positions:
                     continue

sanguo_portfolio/strategies/all_weather_ex.py

@@ -29,6 +29,7 @@ import numpy as np
 import pandas as pd
 
 from .. import factors, filters
+from .all_weather import _total_value  # 2026-08-26 统一 _ex 定寸口径
 
 logger = logging.getLogger(__name__)
 
@@ -204,7 +205,10 @@ class AllWeatherExStrategy:
             # 跌幅最负的 n_pick 只(即"补跌最多")
             idx_sorted = np.argsort(drops)[:n_pick]
             picked = [remaining[i] for i in idx_sorted]
-            cash = _available_cash(context) / n_pick
+            total = _total_value(context)
+            if total <= 0:
+                return
+            cash = total / n_pick
             for code in picked:
                 self.broker.order_value(code, cash)
                 logger.debug("补跌最多的N支 Order %s", code)
@@ -273,7 +277,10 @@ class AllWeatherExStrategy:
         position_count = len(positions)
         target_num = len(target)
         if target_num > position_count:
-            cash = _available_cash(context) / (target_num - position_count)
+            total = _total_value(context)
+            if total <= 0:
+                return
+            cash = total / (target_num - position_count)
             for stock in target:
                 if stock in positions:
                     continue

sanguo_portfolio/strategies/momentum_timing.py

@@ -34,10 +34,10 @@ import pandas as pd
 from .. import filters
 from .all_weather import (
     BrokerFacade,
-    _available_cash,
     _current_dt,
     _dedup,
     _get_positions,
+    _total_value,
 )
 
 logger = logging.getLogger(__name__)
@@ -226,10 +226,10 @@ class MomentumTimingStrategy:
         target_num = len(stocks)
         if target_num == 0:
             return
-        cash = _available_cash(context)
-        if cash <= 0:
+        total = _total_value(context)
+        if total <= 0:
             return
-        per_value = cash / target_num
+        per_value = total / target_num
         for stock in stocks:
             if stock in positions:
                 continue

sanguo_portfolio/strategies/small_cap.py

@@ -44,6 +44,7 @@ from .all_weather import (
     _current_dt,
     _dedup,
     _get_positions,
+    _total_value,
     _previous_date_str,
 )
 
@@ -382,10 +383,10 @@ class SmallCapStrategy:
         # 2) 等额买入 target 中的新股(原策略 per_value = stock_value/len)
         positions = _get_positions(context, self.broker)  # 刷新
         target_num = len(target_stocks)
-        cash = _available_cash(context)
-        if cash <= 0 or target_num == 0:
+        total = _total_value(context)
+        if total <= 0 or target_num == 0:
             return
-        per_value = cash / target_num
+        per_value = total / target_num
         for code in target_stocks:
             if code in positions:
                 continue

sanguo_portfolio/strategies/value_selection.py

@@ -45,6 +45,7 @@ from .all_weather import (
     _current_dt,
     _dedup,
     _get_positions,
+    _total_value,
     _previous_date_str,
 )
 
@@ -197,10 +198,10 @@ class ValueSelectionStrategy:
         target_num = len(buy_list)
         if target_num == 0:
             return
-        cash = _available_cash(context)
-        if cash <= 0:
+        total = _total_value(context)
+        if total <= 0:
             return
-        per_value = cash / target_num
+        per_value = total / target_num
         for stock in buy_list:
             if stock in positions:
                 continue

sanguo_portfolio/strategies/value_selection_ex.py

@@ -285,7 +285,7 @@ class ValueSelectionExStrategy:
             cfg.earnings_growth_low, cfg.earnings_growth_high,
         )
 
-        out = list(l1 & l2 & l3 & l4 & l5 & l6)
+        out = sorted(l1 & l2 & l3 & l4 & l5 & l6)
         logger.info(
             "[%s] L1=%d L2=%d L3=%d L4=%d L5=%d L6=%d → final=%d",
             date_str, len(l1), len(l2), len(l3), len(l4), len(l5), len(l6),

tests/portfolio/test_live_reconcile.py

@@ -33,8 +33,13 @@ def _order(oid="o1", broker_oid="1001", security="000049.XSHE", is_buy=True,
     )
 
 
+_TODAY = datetime.now().strftime("%Y-%m-%d")
+
+
 def _qmt_trade(order_id="1001", security="000049.XSHE", amount=500, price=15.0,
-               trade_id="90001", time="2026-08-25 09:36:24", commission=0.0, tax=0.0):
+               trade_id="90001", time=None, commission=0.0, tax=0.0):
+    """time 缺省=动态今天(EOD 对账按当日过滤,写死日期跨日必翻红,08-26 实锤)。"""
+    time = time or f"{_TODAY} 09:36:24"
     return {
         "trade_id": trade_id, "order_id": order_id, "security": security,
         "amount": amount, "price": price, "time": time,
@@ -199,7 +204,7 @@ class TestEodReconcile:
         save_trade(db, 20, {
             "strategy_name": "s", "symbol": "000049.XSHE",
             "direction": "buy", "offset": "open", "price": 15.0,
-            "volume": 500, "traded_at": "2026-08-25 09:36:24",
+            "volume": 500, "traded_at": f"{_TODAY} 09:36:24",
             "vt_tradeid": "90001"})
         led = LiveInstanceLedger(initial_cash=100_000)
         led.apply_trade(True, "000049.XSHE", 15.0, 500, "90001", "2026-08-25")
@@ -215,7 +220,7 @@ class TestEodReconcile:
         save_trade(db, 20, {
             "strategy_name": "s", "symbol": "000049.XSHE",
             "direction": "buy", "offset": "open", "price": 15.0,
-            "volume": 500, "traded_at": "2026-08-25 09:36:24",
+            "volume": 500, "traded_at": f"{_TODAY} 09:36:24",
             "vt_tradeid": ""})
         led = LiveInstanceLedger(initial_cash=100_000)
         eng = _engine([_order(status="filled", filled=500)], [_qmt_trade()])
@@ -285,7 +290,7 @@ class TestIncidentReplay:
         # 09:36:24 迟到 fill:只在 QMT 原始行里(engine.get_trades 见不到)
         eng = _engine(
             [_order(status="filled", filled=500)],
-            [_qmt_trade(time="2026-08-25 09:36:24", amount=500, price=15.0)])
+            [_qmt_trade(amount=500, price=15.0)])
         assert reconcile_pending(eng, led, db, 19, "momentum_timing") == 1
         # 名单出清 + EOD 复核:无缺口、无重复
         summary = eod_reconcile(eng, led, db, 19, "momentum_timing")

tests/portfolio/test_sizing_total_value.py

@@ -1,14 +1,16 @@
-"""定寸总资产口径(A 修法,前后端 session 2026-08-25 移交)。
+"""定寸总资产口径(A 修法 2026-08-25 → 全量统一 2026-08-26)。
 
 08-25 9:30 事故:small_cap 同轮「全卖19只→马上全买20只」,台账现金只被
 runner_live 的 60s 归因轮询更新(唯一 cash 更新入口)→ 买入定寸在两轮轮询
 之间评估 → 卖出回款不可见 → 20 笔买入全部目标 0,全天空仓。B 修法
-(191270c,下单返回即时归因入账)已做;本组测试钉 **A 修法=防御纵深第二层**:
-定寸口径回总资产 per_value = 总资产/N(现金+持仓市值,卖出前后不变),
-即使台账归因滞后/异常,定寸也不再依赖未入账的现金。
+(191270c,下单返回即时归因入账)已做;A 修法=防御纵深第二层:定寸口径回
+总资产 per_value = 总资产/N(现金+持仓市值,卖出前后不变),即使台账归因
+滞后/异常,定寸也不再依赖未入账的现金。
 
-原 JQ 语义 ``per_value = stock_value/len`` 本就是总资产口径——cash 口径是
-移植时引入的失真,本修法=回原语义。守卫同步 cash<=0 → total<=0。
+**2026-08-26 用户拍板「都要统一ex的」**:原版(回测)四策略定寸同步换
+_total_value——原 JQ 语义 ``per_value = stock_value/len`` 本就是总资产口径,
+cash 口径是移植时引入的失真;统一后回测↔实盘对齐,原版与 _ex 副本同参
+选股+定寸一致(本文件参数化跑 原版+_ex 两类即验收)。守卫 cash<=0 → total<=0。
 """
 from __future__ import annotations
 
@@ -20,25 +22,35 @@ from unittest.mock import MagicMock
 
 from tests.portfolio.conftest import FakeContext, FakePosition
 from sanguo_portfolio.strategies import (
+    AllWeatherExStrategy,
+    AllWeatherStrategy,
     MomentumTimingExStrategy,
+    MomentumTimingStrategy,
     SmallCapExStrategy,
+    SmallCapStrategy,
     ValueSelectionExStrategy,
+    ValueSelectionStrategy,
 )
 from sanguo_portfolio.strategies.all_weather import _total_value
 
 
 # =================== 公共装配 ===================
 class _RecordingBroker:
-    """记录 order_target_value 委托;**不动 context**——精确模拟「卖出已成交、
-    台账归因未跑」的现金窗口(事故形态)。"""
+    """记录 order_target_value / order_value 委托;**不动 context**——精确模拟
+    「卖出已成交、台账归因未跑」的现金窗口(事故形态)。"""
 
     def __init__(self) -> None:
         self.orders: List[Tuple[str, float]] = []
+        self.values: List[Tuple[str, float]] = []
 
     def order_target_value(self, code: str, value: float) -> Any:
         self.orders.append((code, float(value)))
         return object()  # 非 None = 下单成功
 
+    def order_value(self, code: str, value: float) -> Any:
+        self.values.append((code, float(value)))
+        return object()
+
 
 def _stale_cash_ctx(total: float = 10500.0, stale_cash: float = 1000.0) -> FakeContext:
     """旧持仓 OLD_A(市值9000)/OLD_B(市值500);现金=回款未入账的旧值 1000。
@@ -60,6 +72,11 @@ def _stale_cash_ctx(total: float = 10500.0, stale_cash: float = 1000.0) -> FakeC
     return ctx
 
 
+def _falsy_status(stocks):
+    return {s: {"is_limit_up": False, "is_limit_down": False, "is_paused": False}
+            for s in stocks}
+
+
 def _assert_sized_by_total(broker: _RecordingBroker) -> None:
     """卖出旧仓两只 + 新仓三只各按 总资产/3=3500 定寸(绝非现金/N≈333)。
 
@@ -91,33 +108,109 @@ class TestTotalValueHelper:
         assert _total_value(ctx) == pytest.approx(1900.0)
 
 
-# =================== 策略层:卖出未入账时定寸不缩水 ===================
-def test_momentum_rotate_sizes_by_total_value():
-    """momentum _rotate_positions:总资产定寸,对未入账卖出免疫。"""
+# =================== 轮动定寸:原版+_ex 同参一致(2026-08-26 统一验收) ===================
+def _run_momentum_rotate(strategy, ctx) -> None:
+    """统一入口:_ex 直接打 _rotate_positions;原版定寸内联 handle_data,
+    数据层全打桩走通到调仓尾。"""
+    stocks = ["NEW_1.XSHG", "NEW_2.XSHG", "NEW_3.XSHG"]
+    if hasattr(strategy, "_rotate_positions"):
+        strategy._rotate_positions(ctx, stocks)
+        return
+    strategy._cal_buy_sign = lambda *a, **k: True
+    strategy._stock_pool_cached = lambda idx, d: ["I1.XSHG"]
+    strategy._ensure_day_panel = lambda *a, **k: None
+    strategy._find_stock_pool = lambda *a, **k: list(stocks)
+    strategy._select_stocks = lambda *a, **k: list(stocks)
+    strategy._get_limit_status = lambda ss, d: _falsy_status(ss)
+    strategy.handle_data(ctx)
+
+
+@pytest.mark.parametrize("cls", [MomentumTimingStrategy, MomentumTimingExStrategy],
+                         ids=["原版", "ex"])
+def test_momentum_rotate_sizes_by_total_value(cls):
+    """momentum 调仓:总资产定寸,对未入账卖出免疫(原版与 _ex 同断言)。"""
     broker = _RecordingBroker()
-    strat = MomentumTimingExStrategy(provider=MagicMock(), broker=broker)
-    strat._rotate_positions(_stale_cash_ctx(), ["NEW_1.XSHG", "NEW_2.XSHG", "NEW_3.XSHG"])
+    strat = cls(provider=MagicMock(), broker=broker)
+    _run_momentum_rotate(strat, _stale_cash_ctx())
     _assert_sized_by_total(broker)
 
 
-def test_small_cap_rebalance_sizes_by_total_value():
-    """small_cap _rebalance:同型回归。"""
+@pytest.mark.parametrize("cls", [SmallCapStrategy, SmallCapExStrategy],
+                         ids=["原版", "ex"])
+def test_small_cap_rebalance_sizes_by_total_value(cls):
+    """small_cap _rebalance:同型回归(原版与 _ex 同断言)。"""
     broker = _RecordingBroker()
-    strat = SmallCapExStrategy(provider=MagicMock(), broker=broker)
+    strat = cls(provider=MagicMock(), broker=broker)
     strat.in_position_stocks = ["NEW_1.XSHG", "NEW_2.XSHG", "NEW_3.XSHG"]
     strat._rebalance(_stale_cash_ctx())
     _assert_sized_by_total(broker)
 
 
-def test_value_monthly_adjustment_sizes_by_total_value():
+@pytest.mark.parametrize("cls", [ValueSelectionStrategy, ValueSelectionExStrategy],
+                         ids=["原版", "ex"])
+def test_value_monthly_adjustment_sizes_by_total_value(cls):
     """value monthly_adjustment:同型回归(选股段打桩,只验定寸腿)。"""
     broker = _RecordingBroker()
-    strat = ValueSelectionExStrategy(provider=MagicMock(), broker=broker)
+    strat = cls(provider=MagicMock(), broker=broker)
     strat._stock_pool = lambda *a, **k: ["C1.XSHG", "C2.XSHG"]
     strat._get_stock_list = lambda *a, **k: ["NEW_1.XSHG", "NEW_2.XSHG", "NEW_3.XSHG"]
-    strat._get_limit_status = lambda stocks, date: {
-        s: {"is_limit_up": False, "is_limit_down": False, "is_paused": False}
-        for s in stocks
-    }
+    strat._get_limit_status = lambda stocks, date: _falsy_status(stocks)
     strat.monthly_adjustment(_stale_cash_ctx())
     _assert_sized_by_total(broker)
+
+
+# =================== all_weather 两现场(2026-08-26 统一补齐) ===================
+@pytest.mark.parametrize("cls", [AllWeatherStrategy, AllWeatherExStrategy],
+                         ids=["原版", "ex"])
+def test_all_weather_stop_loss_adds_by_total_value(cls):
+    """stop_loss 补跌加仓:order_value = 总资产/n_pick(非 现金/n_pick)。
+
+    形态:STOP 触发 -8% 止损(num_sold=1)→ HOLD 补跌加仓 1 只;
+    n_pick=1 → 加仓额=总资产 10500,现金口径只会下 1000。
+    """
+    broker = _RecordingBroker()
+    strat = cls(provider=MagicMock(), broker=broker)
+    strat.yesterday_hl_list = []
+    strat.config.stop_loss_pct = 0.92
+    strat.config.stock_num = 5
+    ctx = FakeContext(
+        current_dt=datetime(2026, 8, 26, 14, 0),
+        positions={
+            "STOP.XSHG": FakePosition("STOP.XSHG", avg_cost=10.0, price=8.9, total=100),
+            "HOLD.XSHG": FakePosition("HOLD.XSHG", avg_cost=10.0, price=9.7, total=100),
+        },
+        cash=1000.0,
+    )
+    ctx.portfolio.total_value = 10500.0
+    strat.stop_loss(ctx)
+    assert broker.orders == [("STOP.XSHG", 0)]  # 止损卖出
+    assert broker.values == [("HOLD.XSHG", pytest.approx(10500.0))]
+    assert broker.values[0][1] > 1000.0  # 现金口径回归红线
+
+
+@pytest.mark.parametrize("cls", [AllWeatherStrategy, AllWeatherExStrategy],
+                         ids=["原版", "ex"])
+def test_all_weather_monthly_spreads_total_over_empty_slots(cls):
+    """monthly_adjustment 买入腿:每股 = 总资产/空余槽位(非 现金/空余槽位)。
+
+    旧仓 2 只(卖出后视图滞后仍计 2)+ 目标 3 只 → 空余 1 槽 → 每只 10500,
+    现金口径只会下 1000。
+    """
+    broker = _RecordingBroker()
+    strat = cls(provider=MagicMock(), broker=broker)
+    strat.yesterday_hl_list = []
+    strat.config.trend_threshold = 0.01
+    strat._stock_pool = lambda idx, d: ["B1.XSHG"] if "000300" in idx else ["S1.XSHG"]
+    strat._market_cap_top = lambda stocks, d, top, n: list(stocks)
+    strat._trend_mean = lambda lst, d, w: 1.0 if lst and "B1" in lst[0] else 0.5
+    strat._pick_big_universe = lambda b, cur, prev: [
+        "NEW_1.XSHG", "NEW_2.XSHG", "NEW_3.XSHG"]
+    strat._get_limit_status = lambda stocks, date: _falsy_status(stocks)
+    strat.monthly_adjustment(_stale_cash_ctx())
+    sells = {c for c, v in broker.orders if v == 0}
+    buys = [(c, v) for c, v in broker.orders if v > 0]
+    assert sells == {"OLD_A.XSHG", "OLD_B.XSHG"}
+    assert [c for c, _ in buys] == ["NEW_1.XSHG", "NEW_2.XSHG", "NEW_3.XSHG"]
+    for _, v in buys:
+        assert v == pytest.approx(10500.0)  # 总资产/(3-2 空余槽)
+        assert v > 1000.0