test_v3_architecture.py

import tempfile
import unittest
from copy import deepcopy
from pathlib import Path

import pandas as pd

from v3_alert_formatter import build_alert
from v3_candlestick_interpreter import interpret_candlesticks
from v3_clawby_router import route_for_robot
from v3_crypto_group_engine import evaluate_crypto_robot
from v3_forward_ledger import append_event, replace_event_state
from v3_event_wakeup import build_wakeup
from v3_group_engine import evaluate_us_robot
from v3_portfolio_engine import build_virtual_subaccounts, construct_portfolio
from v3_robot_registry import RobotDefinition, build_status, load_registry
from v3_signal_funnel import build_signal_funnel
from v3_shadow_runtime import (
    _clawby_cache_key,
    _clawby_refresh_required,
    enforce_strategy_revision_epoch,
    load_strategy,
)
from v3_validation_adapter import resolve_robot_validation
from us_equity_factor_library import (
    INSTITUTIONAL_FORMULA_FACTOR_COUNT,
    US_EQUITY_BASE_FACTOR_COUNT,
    US_EQUITY_EFFECTIVE_FACTOR_COUNT,
    US_EQUITY_FACTOR_CATALOG,
    apply_us_equity_factor_filter,
)


class V3ArchitectureTests(unittest.TestCase):
    def test_strategy_kernels_are_owned_by_v3_product(self):
        for strategy_id in map(str, range(1, 11)):
            module = load_strategy(strategy_id)
            self.assertEqual(Path(module.__file__).parent.name, "v3_strategy_kernels")

    def test_validation_adapter_fails_closed_without_frozen_score(self):
        result = resolve_robot_validation("G1-US01", "15m")
        self.assertIsNone(result["score"])
        self.assertFalse(result["score_available"])
        self.assertFalse(result["walk_forward_passed"])

    def test_registry_has_five_groups_and_unique_factor_stacks(self):
        status = build_status()
        self.assertEqual(status["group_count"], 5)
        self.assertEqual(status["robot_count"], 55)
        self.assertEqual(status["us_robot_count"], 40)
        self.assertEqual(status["crypto_robot_count"], 10)
        self.assertEqual(status["reviewer_count"], 5)
        self.assertEqual(status["unique_trading_factor_combinations"], 50)
        self.assertEqual(status["candlestick_score_contribution"], 0.0)
        _, robots = load_registry()
        self.assertTrue(all(3 <= len(row.factors) <= 6 for row in robots if row.trading_allowed))

    def test_formula_extension_is_additive_unvalidated_and_non_promoting(self):
        self.assertEqual(US_EQUITY_BASE_FACTOR_COUNT, 200)
        self.assertEqual(INSTITUTIONAL_FORMULA_FACTOR_COUNT, 25)
        self.assertEqual(US_EQUITY_EFFECTIVE_FACTOR_COUNT, 225)
        institutional = [
            spec for spec in US_EQUITY_FACTOR_CATALOG.values()
            if spec.get("family") == "institutional_formula_v3"
        ]
        self.assertEqual(len(institutional), 25)
        self.assertTrue(all(spec.get("automatic_activation_allowed") is False for spec in institutional))
        self.assertTrue(all("unvalidated" in str(spec.get("historical_backtest_status")) or "research_only" in str(spec.get("historical_backtest_status")) for spec in institutional))

    def test_completed_bar_formula_factors_gate_by_direction(self):
        count = 90
        close = pd.Series([120 - index * 0.2 for index in range(count)], dtype="float64")
        volume = pd.Series([1_000_000 + index * 1_000 for index in range(count)], dtype="float64")
        volume.iloc[-1] = 2_000_000
        frame = pd.DataFrame({
            "open": close - 0.1,
            "high": close + 0.4,
            "low": close - 0.4,
            "close": close,
            "volume": volume,
        })
        long_signal = pd.Series(1, index=frame.index)
        short_signal = pd.Series(-1, index=frame.index)
        reversal_long = apply_us_equity_factor_filter(long_signal, frame, "volume_gated_7d_reversal", timeframe="1d")
        reversal_short = apply_us_equity_factor_filter(short_signal, frame, "volume_gated_7d_reversal", timeframe="1d")
        range_long = apply_us_equity_factor_filter(long_signal, frame, "range_momentum")
        self.assertEqual(int(reversal_long.iloc[-1]), 1)
        self.assertEqual(int(reversal_short.iloc[-1]), 0)
        self.assertEqual(int(range_long.iloc[-1]), 1)
        self.assertIsNone(apply_us_equity_factor_filter(long_signal, frame, "volume_gated_7d_reversal", timeframe="15m"))

    def test_context_formula_never_falls_back_to_ohlcv(self):
        _, robots = load_registry()
        robot = next(row for row in robots if row.robot_id == "G4-US01")
        count = 80
        close = pd.Series([100 + index * 0.05 for index in range(count)])
        frame = pd.DataFrame({
            "open": close.shift(1).fillna(close.iloc[0]), "high": close + 0.5, "low": close - 0.5,
            "close": close, "volume": [1000 + index for index in range(count)],
            "bar_close_time": pd.date_range("2026-07-13T13:30:00Z", periods=count, freq="15min"),
        })
        frame["available_time"] = frame["bar_close_time"] + pd.Timedelta(seconds=1)
        result = evaluate_us_robot(
            robot, frame, upstream_signal=1, upstream_score=84,
            data_gate_passed=True, walk_forward_passed=True,
            factor_context={}, decision_time=frame["available_time"].iloc[-1],
        )
        self.assertIn("factor_unavailable:residual_medium_momentum", result["blockers"])
        row = next(item for item in result["factor_evidence"] if item["factor_id"] == "residual_medium_momentum")
        self.assertEqual(row["status"], "UNAVAILABLE")

    def test_candlestick_only_after_qualified_signal_and_never_scores(self):
        frame = pd.DataFrame({
            "open": [10.5, 10.3, 10.1, 10.0, 9.9, 9.7],
            "high": [10.6, 10.4, 10.2, 10.1, 10.0, 10.3],
            "low": [10.2, 10.0, 9.8, 9.7, 9.6, 9.65],
            "close": [10.3, 10.1, 9.9, 9.8, 9.7, 10.2],
            "bar_close_time": pd.date_range("2026-07-14T13:00:00Z", periods=6, freq="15min"),
            "available_time": pd.date_range("2026-07-14T13:00:01Z", periods=6, freq="15min"),
        })
        blocked = interpret_candlesticks(frame, timeframe="15m", qualified_score=69.9, hard_gates_passed=True)
        self.assertEqual(blocked["status"], "NOT_EVALUATED")
        result = interpret_candlesticks(
            frame, timeframe="15m", decision_time="2026-07-14T14:15:02Z",
            qualified_score=82.0, hard_gates_passed=True,
        )
        self.assertEqual(result["score_contribution"], 0.0)
        self.assertFalse(result["may_change_score"])
        self.assertIn("bullish_engulfing", {row["pattern_id"] for row in result["patterns"]})

    def test_clawby_router_is_mechanism_aware_without_signal_authority(self):
        robot = {
            "robot_id": "G3-US01", "market": "US_EQUITY",
            "factor_roles": {"core_alpha": ["return_momentum_w5"], "confirmation": ["clawby_dark_pool_flow"], "risk": ["gap_fade_guard"]},
        }
        quiet = route_for_robot(robot, raw_signal_present=False)
        active = route_for_robot(robot, event_types=["earnings"], raw_signal_present=True)
        self.assertFalse(quiet["clawby_signal_authority"])
        self.assertNotIn("clawby:dark_pool", quiet["calls"])
        self.assertIn("clawby:dark_pool", active["calls"])
        self.assertIn("clawby:options", active["calls"])
        self.assertLessEqual(len(active["calls"]), 12)

    def test_crypto_robot_requires_pit_factor_context(self):
        _, robots = load_registry()
        robot = next(row for row in robots if row.robot_id == "G1-CR01")
        context = {
            factor_id: {"passed": True, "hard_veto": False, "available_time": "2026-07-14T14:00:00Z"}
            for factor_id in robot.factors
        }
        result = evaluate_crypto_robot(
            robot, upstream_direction="LONG", upstream_score=84,
            decision_time="2026-07-14T14:00:01Z", factor_context=context,
            data_gate_passed=True, walk_forward_passed=True,
        )
        self.assertTrue(result["qualified"])
        self.assertEqual(result["final_score"], 84)
        context[robot.factors[0]]["available_time"] = "2026-07-14T14:00:02Z"
        blocked = evaluate_crypto_robot(
            robot, upstream_direction="LONG", upstream_score=84,
            decision_time="2026-07-14T14:00:01Z", factor_context=context,
            data_gate_passed=True, walk_forward_passed=True,
        )
        self.assertFalse(blocked["qualified"])

    def test_robot_evidence_is_independent_but_portfolio_trade_is_deduplicated(self):
        candidates = []
        for robot_id in ("G1-US01", "G2-US04"):
            candidates.append({
                "qualified": True, "robot_id": robot_id, "group_id": robot_id[:2],
                "market": "US_EQUITY", "ticker": "IBM", "sector": "Technology",
                "timeframe": "15m", "signal_bar_close": "2026-07-14T14:00:00+00:00",
                "direction": "SHORT", "final_score": 84.0, "reliability": 0.7,
                "annualized_volatility": 0.3,
            })
        subaccounts = build_virtual_subaccounts(candidates)
        portfolio = construct_portfolio(subaccounts)
        self.assertEqual(len(subaccounts), 2)
        self.assertEqual(portfolio["portfolio_trade_count"], 1)
        self.assertEqual(portfolio["trades"][0]["robot_evidence_count"], 2)
        self.assertFalse(portfolio["automatic_order_allowed"])

    def test_cross_timeframe_instrument_exposure_cannot_exceed_five_percent(self):
        candidates = []
        for timeframe, signal_bar_close in (
            ("15m", "2026-07-14T14:00:00Z"),
            ("1h", "2026-07-14T14:30:00Z"),
            ("4h", "2026-07-14T17:30:00Z"),
        ):
            candidates.append({
                "qualified": True, "robot_id": f"G1-US01-{timeframe}", "group_id": "G1",
                "market": "US_EQUITY", "ticker": "IBM", "sector": "Technology",
                "timeframe": timeframe, "signal_bar_close": signal_bar_close,
                "direction": "LONG", "final_score": 90.0, "reliability": 1.0,
                "annualized_volatility": 0.10,
            })
        portfolio = construct_portfolio(build_virtual_subaccounts(candidates))
        ibm_exposure = sum(
            float(row["position_fraction"]) for row in portfolio["trades"] if row["ticker"] == "IBM"
        )
        self.assertLessEqual(ibm_exposure, 0.05)
        self.assertTrue(any(
            row["status"].startswith("INSTRUMENT_EXPOSURE_LIMIT") for row in portfolio["conflicts"]
        ))

    def test_event_wakeup_and_signal_funnel_are_semantically_separate(self):
        wakeup = build_wakeup({"event_type": "earnings", "ticker": "IBM", "available_time": "2026-07-14T13:30:00Z"})
        self.assertTrue(wakeup["wakeup"])
        self.assertFalse(wakeup["creates_direction"])
        funnel = build_signal_funnel(
            wakeups=[wakeup],
            evaluations=[{"direction": "LONG", "final_score": 69, "blockers": ["score_below_alert_threshold"], "qualified": False}],
        )
        stages = {row["stage"]: row["count"] for row in funnel["stages"]}
        self.assertEqual(stages["market_anomaly"], 1)
        self.assertEqual(stages["raw_signal"], 1)
        self.assertEqual(stages["qualified_robot_signal"], 0)

    def test_signal_funnel_handles_unavailable_score(self):
        funnel = build_signal_funnel(evaluations=[{
            "direction": "LONG", "final_score": None,
            "blockers": ["score_unavailable"], "qualified": False,
        }])
        stages = {row["stage"]: row["count"] for row in funnel["stages"]}
        self.assertEqual(stages["raw_signal"], 1)
        self.assertEqual(stages["score_threshold_pass"], 0)

    def test_signal_funnel_preserves_raw_signal_blocked_by_factors(self):
        funnel = build_signal_funnel(evaluations=[{
            "raw_direction": "LONG",
            "direction": "FLAT",
            "final_score": None,
            "qualified": False,
            "blockers": ["factor_stack_neutral_or_blocked"],
        }])
        stages = {row["stage"]: row["count"] for row in funnel["stages"]}
        self.assertEqual(stages["raw_signal"], 1)
        self.assertEqual(stages["qualified_robot_signal"], 0)

    def test_us_robot_ignores_future_external_factor_context(self):
        robot = RobotDefinition(
            robot_id="TEST-US", group_id="TEST", group_label="test", slot="US01",
            market="US_EQUITY", role="trading", base_strategy="1",
            core_alpha=("ema_alignment_f5_s21",),
            confirmation=("clawby_dark_pool_flow",), risk=(), trading_allowed=True,
        )
        count = 80
        close = pd.Series([100 + index * 0.05 for index in range(count)])
        frame = pd.DataFrame({
            "open": close.shift(1).fillna(close.iloc[0]), "high": close + 0.5, "low": close - 0.5,
            "close": close, "volume": [1000 + index for index in range(count)],
            "bar_close_time": pd.date_range("2026-07-13T13:30:00Z", periods=count, freq="15min"),
        })
        frame["available_time"] = frame["bar_close_time"] + pd.Timedelta(seconds=1)
        context = {"clawby_dark_pool_flow": {
            "passed": True, "hard_veto": False, "available_time": "2026-07-14T14:00:02Z",
        }}
        result = evaluate_us_robot(
            robot, frame, upstream_signal=1, upstream_score=84,
            data_gate_passed=True, walk_forward_passed=True,
            factor_context=context, decision_time="2026-07-14T14:00:01Z",
        )
        self.assertNotIn("factor_blocked:clawby_dark_pool_flow", result["blockers"])
        self.assertEqual(result["final_score"], 84)
        self.assertTrue(result["formal_policy_eligible"])
        clawby = next(
            row for row in result["factor_evidence"]
            if row["factor_id"] == "clawby_dark_pool_flow"
        )
        self.assertTrue(clawby["advisory_only"])
        self.assertEqual(clawby["score_contribution"], 0.0)
        self.assertFalse(clawby["may_block_signal"])

    def test_clawby_refresh_cache_is_namespaced_by_requested_routes(self):
        dark_pool = _clawby_cache_key(
            "IBM", "LONG", {"calls": ["clawby:dark_pool"]}
        )
        options = _clawby_cache_key(
            "IBM", "LONG", {"calls": ["clawby:options"]}
        )
        reordered = _clawby_cache_key(
            "IBM",
            "LONG",
            {"calls": ["clawby:options", "clawby:dark_pool"]},
        )
        sorted_routes = _clawby_cache_key(
            "IBM",
            "LONG",
            {"calls": ["clawby:dark_pool", "clawby:options"]},
        )
        self.assertNotEqual(dark_pool, options)
        self.assertEqual(reordered, sorted_routes)

    def test_clawby_refreshes_after_qualification_not_for_every_raw_signal(self):
        raw_only = _clawby_refresh_required(
            direction="LONG",
            result={"qualified": False, "adaptive_paper_qualified": False},
            needs_clawby=True,
        )
        qualified = _clawby_refresh_required(
            direction="SHORT",
            result={"qualified": True},
            needs_clawby=True,
        )
        event_review = _clawby_refresh_required(
            direction="LONG",
            result={"qualified": False},
            needs_clawby=True,
            event_count=1,
        )
        self.assertFalse(raw_only)
        self.assertTrue(qualified)
        self.assertTrue(event_review)

    def test_strategy_revision_epoch_rejects_pre_fix_bars(self):
        candidate = {
            "qualified": True,
            "formal_policy_eligible": True,
            "adaptive_paper_qualified": True,
            "blockers": [],
            "strategy_revision_effective_from": "2026-07-23T05:42:03Z",
        }
        blocked = enforce_strategy_revision_epoch(
            deepcopy(candidate), decision_time="2026-07-23T05:42:02Z"
        )
        self.assertFalse(blocked["qualified"])
        self.assertFalse(blocked["formal_policy_eligible"])
        self.assertFalse(blocked["adaptive_paper_qualified"])
        self.assertIn(
            "strategy_revision_waiting_for_first_post_fix_bar",
            blocked["blockers"],
        )
        allowed = enforce_strategy_revision_epoch(
            deepcopy(candidate), decision_time="2026-07-23T05:42:04Z"
        )
        self.assertTrue(allowed["qualified"])

    def test_forward_ledger_is_idempotent_and_immutable(self):
        event = {
            "event_id": "e1", "run_id": "r1", "market": "US_EQUITY", "ticker": "IBM",
            "robot_id": "G1-US01", "group_id": "G1", "timeframe": "15m", "direction": "SHORT",
            "signal_bar_close": "2026-07-14T14:00:00+00:00", "decision_time": "2026-07-14T14:00:02+00:00",
            "execution_status": "pending_next_executable_quote", "paper_trading_only": True,
        }
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "ledger.jsonl"
            self.assertEqual(append_event(event, path)["status"], "APPENDED")
            self.assertEqual(append_event(event, path)["status"], "ALREADY_PRESENT")
            updated = dict(event, execution_status="captured", entry_price=100.0)
            self.assertEqual(replace_event_state(updated, path)["status"], "UPDATED")
            with self.assertRaises(RuntimeError):
                replace_event_state(dict(updated, ticker="QCOM"), path)

    def test_alert_requires_tp_sl_time_and_zero_score_candlestick(self):
        event = {
            "paper_trading_only": True, "execution_status": "captured", "portfolio_trade_identity": "p1",
            "direction": "LONG", "entry_price": 100.0, "score": 82.0, "blockers": [],
            "ticker": "IBM", "market": "US_EQUITY", "timeframe": "15m", "group_id": "G1",
            "robot_id": "G1-US01", "signal_bar_close": "2026-07-14T14:00:00Z",
            "decision_time": "2026-07-14T14:00:02Z", "execution_time": "2026-07-14T14:00:03Z",
        }
        trade = {"paper_trading_only": True, "portfolio_trade_id": "p1", "position_fraction": 0.012}
        reviewer = {
            "candlestick_score_contribution": 0.0,
            "candlestick": {"score_contribution": 0.0, "dominant_pattern": {
                "pattern_name": "锤子线",
                "common_interpretation": "卖压被承接",
                "confirmation": "下一根已完成K线突破形态高点",
                "invalidation": "跌破形态低点",
            }},
        }
        alert = build_alert(event, trade, reviewer, take_profit=108, stop_loss=96, invalidation="跌破96", actual_time="2026-07-14T14:00:04Z")
        self.assertIn("止盈：108.0000", alert["message_zh"])
        self.assertIn("评分贡献0", alert["message_zh"])
        self.assertIn("日本蜡烛图：锤子线", alert["message_zh"])
        self.assertIn("常见后续：卖压被承接", alert["message_zh"])
        self.assertIn("形态确认：下一根已完成K线突破形态高点", alert["message_zh"])
        self.assertIn("形态失效：跌破形态低点", alert["message_zh"])
        self.assertIn("不单独触发开仓", alert["message_zh"])
        self.assertEqual(alert["capital_profile"]["code"], "SMALL_PAPER_BUDGET")
        self.assertAlmostEqual(alert["planned_position_fraction"], 0.012)
        self.assertIn("纸盘资金容量：小额试探", alert["message_zh"])
        with self.assertRaises(ValueError):
            build_alert(event, trade, reviewer, take_profit=95, stop_loss=96, invalidation="跌破96", actual_time="2026-07-14T14:00:04Z")


if __name__ == "__main__":
    unittest.main()