test_validation_pipeline.py

import json
import unittest
import tempfile
from pathlib import Path
from unittest import mock

LEGACY_V2_MODULES = {
    "dark_pool_factor.py", "factor_evolution_lab.py", "filter_ablation_experiments.py",
    "institutional_metrics.py", "meta_alpha_engine.py", "point_in_time_feature_collector.py",
    "realtime_alert_scanner.py", "risk_engine.py", "robot_factor_stock_selector.py",
    "robot_validation_summary.py", "robot_council.py", "strategy_governance.py",
    "single_delta_experiments.py", "sentinel_robot.py", "us_equity_factor_library.py",
    "validation_engine.py",
}
LEGACY_V2_ROOT = Path(__file__).resolve().parent
LEGACY_V2_MISSING = sorted(
    name for name in LEGACY_V2_MODULES if not (LEGACY_V2_ROOT / name).exists()
)
if LEGACY_V2_MISSING:
    raise unittest.SkipTest(
        "retired V2 validation archive; missing modules: " + ", ".join(LEGACY_V2_MISSING)
    )

import pandas as pd

from dark_pool_factor import calculate_dark_pool_factor
from common import _clawby_rate_acquire, refresh_clawby_account_rate_limit
from factor_evolution_lab import apply_governance_freeze_to_state
from filter_ablation_experiments import apply_companion_filter
from institutional_metrics import backtest_metrics, backtest_return_series
from meta_alpha_engine import meta_alpha_decision
from point_in_time_feature_collector import assess_strategy_data
from realtime_alert_scanner import _refresh_directional_factor_pack
from risk_engine import adaptive_position_fraction
from robot_factor_stock_selector import factor_assignment
from robot_validation_summary import build_robot_validation_summary
from robot_council import read_backtest_results
from strategy_governance import apply_factor_freeze, build_governance, promotion_decision
from single_delta_experiments import apply_single_delta
from sentinel_robot import consensus_eligibility, market_clock, mode_allowed
from us_equity_factor_library import US_EQUITY_FACTOR_CATALOG, apply_us_equity_factor_filter
from validation_engine import holdout_acceptance, walk_forward_summary


class ValidationPipelineTests(unittest.TestCase):
    def test_funding_and_oi_robots_are_crypto_primary_for_governance(self):
        folds = pd.DataFrame([{"fold": 1, "test_start": "2025-01-01", "test_end": "2025-02-01"}])
        governance = build_governance(pd.DataFrame(), pd.DataFrame(), folds, point_in_time_status={})
        for strategy in ("4", "6"):
            item = governance["robots"][strategy]
            self.assertEqual(item["primary_market"], "CRYPTO_PERPETUAL")
            self.assertEqual(item["us_equity_applicability"], "not_applicable")
            self.assertFalse(item["strategy_promotion_allowed"])
            self.assertIn("us_equity_not_applicable_crypto_primary", item["promotion_blocked_reasons"])

    def test_sentinel_respects_nyse_holidays_and_dst(self):
        holiday = market_clock("2026-07-03 08:30:00-04:00")
        summer = market_clock("2026-07-10 09:35:00-04:00")
        winter = market_clock("2026-01-06 09:35:00-05:00")
        self.assertFalse(holiday["is_session"])
        self.assertFalse(mode_allowed("premarket", holiday)[0])
        self.assertTrue(mode_allowed("intraday", summer)[0])
        self.assertTrue(mode_allowed("intraday", winter)[0])
        self.assertEqual(summer["market_open"].strftime("%H:%M"), "09:30")
        self.assertEqual(winter["market_open"].strftime("%H:%M"), "09:30")

    def test_sentinel_consensus_hard_gate(self):
        self.assertEqual(consensus_eligibility(["1", "2"], 1, 0, True), (True, "ok"))
        self.assertFalse(consensus_eligibility(["1"], 1, 0, True)[0])
        self.assertFalse(consensus_eligibility(["1", "9"], 1, 1, True)[0])
        self.assertFalse(consensus_eligibility(["1", "2"], 1, -1, True)[0])
        self.assertFalse(consensus_eligibility(["1", "2"], 1, 0, False)[0])

    def test_filter_ablation_applies_confirmation_and_veto_without_changing_baseline(self):
        signal = pd.Series([1, 1, -1, -1, 0])
        companion = pd.Series([1, -1, -1, 1, 1])
        baseline = apply_companion_filter(signal, companion, "same_direction", enabled=False)
        confirmed = apply_companion_filter(signal, companion, "same_direction", enabled=True)
        vetoed = apply_companion_filter(signal, companion, "opposite_veto", enabled=True)
        self.assertEqual(baseline.tolist(), signal.tolist())
        self.assertEqual(confirmed.tolist(), [1, 0, -1, 0, 0])
        self.assertEqual(vetoed.tolist(), [1, 0, -1, 0, 0])

    def test_robot_council_reads_explicit_backtest_columns(self):
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "backtests.csv"
            pd.DataFrame(
                [{"strategy": "1", "ticker": "QCOM", "window": "2y", "trades": 40, "unused": "x"}]
            ).to_csv(path, index=False)
            result = read_backtest_results(path)
        self.assertEqual(list(result.columns), ["strategy", "ticker", "window", "trades"])

    def test_clawby_budget_reserves_realtime_capacity(self):
        with tempfile.TemporaryDirectory() as directory:
            state = Path(directory) / "state.json"
            lock = Path(directory) / "state.lock"
            allowed_normal = [
                _clawby_rate_acquire("normal", state, lock, now=1000 + index * 0.01, limit=60, reserve=12)[0]
                for index in range(49)
            ]
            self.assertEqual(sum(allowed_normal), 48)
            allowed_realtime = [
                _clawby_rate_acquire("realtime", state, lock, now=1001 + index * 0.01, limit=60, reserve=12)[0]
                for index in range(13)
            ]
            self.assertEqual(sum(allowed_realtime), 12)

    def test_clawby_research_budget_yields_to_live_traffic(self):
        with tempfile.TemporaryDirectory() as directory:
            state = Path(directory) / "state.json"
            lock = Path(directory) / "state.lock"
            allowed_research = [
                _clawby_rate_acquire("research", state, lock, now=1000 + index * 0.01, limit=60, reserve=12)[0]
                for index in range(31)
            ]
            self.assertEqual(sum(allowed_research), 30)
            allowed_critical = [
                _clawby_rate_acquire("critical", state, lock, now=1001 + index * 0.01, limit=60, reserve=12)[0]
                for index in range(31)
            ]
            self.assertEqual(sum(allowed_critical), 30)

    def test_upgraded_clawby_1200_budget_preserves_scaled_priority_capacity(self):
        with tempfile.TemporaryDirectory() as directory:
            state = Path(directory) / "state.json"
            lock = Path(directory) / "state.lock"
            normal = [
                _clawby_rate_acquire("normal", state, lock, now=1000 + index * 0.001, limit=1200, reserve=240)[0]
                for index in range(961)
            ]
            self.assertEqual(sum(normal), 960)
            realtime = [
                _clawby_rate_acquire("realtime", state, lock, now=1001 + index * 0.001, limit=1200, reserve=240)[0]
                for index in range(241)
            ]
            self.assertEqual(sum(realtime), 240)
            status = json.loads(state.read_text(encoding="utf-8"))
            self.assertEqual(status["priority_ceilings"], {
                "critical": 1200, "realtime": 1200, "normal": 960, "research": 600,
            })

    def test_official_account_limit_can_only_reduce_the_local_ceiling(self):
        with tempfile.TemporaryDirectory() as directory:
            state = Path(directory) / "state.json"
            lock = Path(directory) / "state.lock"
            response = mock.MagicMock()
            response.read.return_value = b'{"rate_per_min": 360}'
            response.__enter__.return_value = response
            with mock.patch("common._clawby_api_key", return_value="test-key"), \
                 mock.patch("common.urllib.request.urlopen", return_value=response):
                result = refresh_clawby_account_rate_limit(state, lock, now=1000, force=True)
            self.assertEqual(result["status"], "OFFICIAL_ACCOUNT_VERIFIED")
            self.assertEqual(result["configured_limit"], 1200)
            self.assertEqual(result["effective_limit"], 360)

    def test_point_in_time_data_does_not_promote_noninformative_or_immature_features(self):
        rows = [
            {
                "ticker": f"T{index}",
                "funding_points": 100,
                "funding_nonzero_points": 0,
                "funding_std": 0.0,
                "prospective_funding_points": 0,
                "prospective_funding_nonzero_points": 0,
                "prospective_funding_std": 0.0,
                "oi_points": 12,
                "oi_span_days": 2.0,
                "hourly_bars": 1000,
                "prospective_hourly_bars": 3,
                "prospective_hourly_span_days": 0.1,
            }
            for index in range(6)
        ]
        funding = assess_strategy_data("4", rows)
        oi = assess_strategy_data("6", rows)
        liquidation = assess_strategy_data("8", rows)
        self.assertEqual(funding["status"], "blocked_non_informative_funding")
        self.assertEqual(oi["status"], "collecting_point_in_time_oi")
        self.assertEqual(liquidation["status"], "collecting_prospective_intraday")
        self.assertFalse(funding["promotion_allowed"])
        self.assertFalse(oi["training_data_ready"])
        self.assertFalse(liquidation["training_data_ready"])

    def test_dark_pool_factor_changes_sign_with_robot_direction(self):
        summary = {
            "total_count": 1000,
            "total_premium": 2_000_000,
            "total_volume": 10_000,
            "ask_premium": 900_000,
            "bid_premium": 300_000,
            "mid_premium": 800_000,
        }
        long_factor = calculate_dark_pool_factor(summary, 1, price=100, average_volume=100_000)
        short_factor = calculate_dark_pool_factor(summary, -1, price=100, average_volume=100_000)
        self.assertTrue(long_factor["available"])
        self.assertGreater(long_factor["factor_adjustment"], 0)
        self.assertLess(short_factor["factor_adjustment"], 0)
        self.assertAlmostEqual(long_factor["factor_adjustment"], -short_factor["factor_adjustment"], places=4)

    def test_dark_pool_factor_missing_data_stays_neutral(self):
        factor = calculate_dark_pool_factor({}, 1, price=100)
        self.assertFalse(factor["available"])
        self.assertEqual(factor["factor_adjustment"], 0.0)

    def test_cached_dark_pool_pack_is_recalculated_for_each_robot_direction(self):
        cached = {
            "ticker": "QCOM",
            "price": 100,
            "sources": ["market_context", "dark_pool"],
            "screener_confirmed": True,
            "ticker_snapshot": {"thirty_day_avg_vol": 100_000, "volume": 80_000},
            "dark_pool_summary": {
                "total_count": 1000,
                "total_premium": 2_000_000,
                "total_volume": 10_000,
                "ask_premium": 900_000,
                "bid_premium": 300_000,
                "mid_premium": 800_000,
            },
        }
        long_pack = _refresh_directional_factor_pack(cached, 1, 100)
        short_pack = _refresh_directional_factor_pack(cached, -1, 100)
        self.assertGreater(long_pack["dark_pool_factor"]["factor_adjustment"], 0)
        self.assertLess(short_pack["dark_pool_factor"]["factor_adjustment"], 0)
        self.assertGreater(long_pack["factor_score"], short_pack["factor_score"])
        self.assertNotIn("dark_pool_factor", cached)

    def test_position_sizing_accepts_pandas_strategy_row(self):
        frame = pd.DataFrame({"close": [100 + index * 0.1 for index in range(40)]})
        sizing = adaptive_position_fraction(
            {"position_fraction": 0.05},
            {
                "final_decision_score": 85,
                "reliability_score": 0.8,
                "liquidity_score": 1.0,
                "instrument_quality_score": 1.0,
            },
            frame,
            signal=1,
            learning={"robot_scoreboard": {"1": {"wins": 2, "losses": 1}}},
            row=pd.Series({"strategy": "1"}),
        )
        self.assertGreater(sizing["position_fraction"], 0)
        self.assertLessEqual(sizing["position_fraction"], 0.05)

    def test_extended_factor_catalog_and_dark_pool_history_boundary(self):
        self.assertEqual(len(US_EQUITY_FACTOR_CATALOG), 225)
        self.assertEqual(
            US_EQUITY_FACTOR_CATALOG["clawby_dark_pool_flow"]["historical_backtest_status"],
            "prospective_only",
        )
        frame = pd.DataFrame(
            {
                "open": [100 + index * 0.2 for index in range(80)],
                "high": [100.4 + index * 0.2 for index in range(80)],
                "low": [99.6 + index * 0.2 for index in range(80)],
                "close": [100 + index * 0.2 for index in range(80)],
                "volume_usd": [1_000_000 + index * 10_000 for index in range(80)],
            }
        )
        signal = pd.Series([1] * len(frame), index=frame.index)
        filtered = apply_us_equity_factor_filter(signal, frame, "ema_alignment_f3_s8")
        self.assertEqual(len(filtered), len(signal))
        self.assertTrue(set(filtered.unique()).issubset({-1, 0, 1}))

    def test_return_path_matches_backtest_equity(self):
        close = pd.Series([100, 101, 99, 102, 103, 100, 104], dtype="float64")
        frame = pd.DataFrame(
            {
                "open": close,
                "high": close + 0.5,
                "low": close - 0.5,
                "close": close,
                "volume_usd": 1_000_000.0,
            }
        )
        signal = pd.Series([0, 1, 1, -1, -1, 0, 0], dtype="int64")
        returns = backtest_return_series(frame, signal, fee_bps=8.0, max_hold_bars=4, stop_loss_pct=-0.04, take_profit_pct=0.08)
        metrics = backtest_metrics(frame, signal, fee_bps=8.0, max_hold_bars=4, stop_loss_pct=-0.04, take_profit_pct=0.08)
        self.assertAlmostEqual(float((1.0 + returns).prod()), metrics["final_equity"], places=10)

    def test_governance_refuses_failed_cross_sectional_candidate(self):
        decision = promotion_decision(
            {
                "cross_sectional_status": "稳定性未通过",
                "oos_trades": 20,
                "oos_ticker_floor": 3,
                "fold_count": 1,
                "oos_institutional_score": 49.9,
                "oos_profit_factor": 1.14,
                "oos_sharpe": 0.1,
                "oos_max_drawdown": -0.10,
                "positive_folds": 1,
            },
            {"specialization_status": "稳定性未通过"},
            "重点研究",
        )
        self.assertFalse(decision["allowed"])
        self.assertIn("oos_trades_below_threshold", decision["reasons"])
        self.assertIn("insufficient_positive_oos_folds", decision["reasons"])

    def test_governance_demotes_unvalidated_active_factors(self):
        learning, demoted = apply_factor_freeze(
            {"factor_evolution": {"robots": {"1": {"active_factors": ["tail_risk_guard"], "watchlist_factors": []}}}},
            {"generated_at": "2026-07-10T00:00:00+00:00", "robots": {"1": {"factor_promotion_allowed": False}}},
        )
        robot = learning["factor_evolution"]["robots"]["1"]
        self.assertEqual(robot["active_factors"], [])
        self.assertEqual(robot["watchlist_factors"], ["tail_risk_guard"])
        self.assertEqual(demoted["1"], ["tail_risk_guard"])

    def test_factor_daemon_state_cannot_restore_frozen_factors(self):
        state = {"updated_at": "2026-07-10T00:00:00+00:00", "robots": {"1": {"active_factors": ["tail_risk_guard"]}}}
        demoted = apply_governance_freeze_to_state(state)
        self.assertEqual(demoted["1"], ["tail_risk_guard"])
        self.assertEqual(state["robots"]["1"]["active_factors"], [])

    def test_gap_single_delta_only_filters_large_gap_signals(self):
        frame = pd.DataFrame(
            {
                "datetime": pd.date_range("2026-01-01", periods=4, tz="UTC"),
                "open": [100.0, 100.0, 105.0, 100.0],
                "high": [101.0, 101.0, 106.0, 101.0],
                "low": [99.0, 99.0, 104.0, 99.0],
                "close": [100.0, 100.0, 105.0, 100.0],
                "volume_usd": [1_000_000.0] * 4,
            }
        )
        filtered, execution = apply_single_delta("1", pd.Series([1, 1, 1, 1]), frame, 0.02, {})
        self.assertEqual(int(filtered.iloc[2]), 0)
        self.assertEqual(execution["max_hold_bars"], 4)

    def test_every_robot_gets_three_to_six_factors_and_dark_pool_rationale(self):
        for strategy in [str(index) for index in range(1, 11)] + ["super", "mix"]:
            assignment = factor_assignment(strategy)
            self.assertGreaterEqual(len(assignment["selected_factors"]), 3)
            self.assertLessEqual(len(assignment["selected_factors"]), 6)
            self.assertTrue(assignment["dark_pool_reason"])

    def test_short_window_small_sample_cannot_replace_stable_long_window(self):
        rows = [
            {
                "strategy": "1",
                "name": "Robot 1",
                "ticker": "FAST",
                "window": "15d",
                "trades": 3,
                "institutional_score": 99,
                "sharpe": 9,
                "effective_sharpe": 2,
                "sample_multiplier": 0.23,
                "walk_forward_pass": True,
                "walk_forward_state": "pass",
                "walk_forward_positive_folds": 3,
                "walk_forward_fold_count": 3,
                "walk_forward_total_trades": 3,
                "applicability_status": "applicable",
            },
            {
                "strategy": "1",
                "name": "Robot 1",
                "ticker": "ROBUST",
                "window": "half_year",
                "trades": 24,
                "institutional_score": 78,
                "sharpe": 2.1,
                "effective_sharpe": 2.1,
                "sample_multiplier": 1,
                "walk_forward_pass": True,
                "walk_forward_state": "pass",
                "walk_forward_positive_folds": 2,
                "walk_forward_fold_count": 3,
                "walk_forward_total_trades": 24,
                "applicability_status": "applicable",
            },
        ]
        result = build_robot_validation_summary(rows).iloc[0]
        self.assertEqual(result["selected_ticker"], "ROBUST")
        self.assertEqual(result["stability_status"], "通过")

    def test_missing_required_data_is_not_reported_as_walk_forward_unknown(self):
        result = build_robot_validation_summary(
            [
                {
                    "strategy": "4",
                    "name": "Robot 4",
                    "ticker": "MSFT",
                    "window": "2y",
                    "trades": 0,
                    "institutional_score": 0,
                    "applicability_status": "missing_required_data",
                    "applicability_reason": "missing funding history",
                    "walk_forward_state": "not_applicable",
                }
            ]
        ).iloc[0]
        self.assertEqual(result["stability_status"], "数据不适用")
        self.assertEqual(result["walk_forward_state"], "not_applicable")

    def test_longer_failed_window_cannot_be_hidden_by_shorter_pass(self):
        common = {
            "strategy": "1",
            "name": "Robot 1",
            "ticker": "QCOM",
            "trades": 24,
            "institutional_score": 80,
            "sharpe": 2,
            "effective_sharpe": 2,
            "sample_multiplier": 1,
            "walk_forward_positive_folds": 2,
            "walk_forward_fold_count": 3,
            "applicability_status": "applicable",
        }
        result = build_robot_validation_summary(
            [
                {**common, "window": "half_year", "walk_forward_state": "pass", "walk_forward_pass": True},
                {**common, "window": "2y", "trades": 60, "walk_forward_state": "fail", "walk_forward_pass": False},
            ]
        ).iloc[0]
        self.assertEqual(result["selected_window"], "2y")
        self.assertEqual(result["stability_status"], "稳定性未通过")

    def test_insufficient_history_has_explicit_state(self):
        frame = pd.DataFrame({"close": range(20)})
        signal = pd.Series(0, index=frame.index)
        result = walk_forward_summary(frame, signal, lambda *_: {})
        self.assertFalse(result["pass"])
        self.assertEqual(result["state"], "insufficient_history")
        self.assertEqual(result["fold_count"], 0)

    def test_holdout_requires_sample_and_profit_factor(self):
        weak = holdout_acceptance(
            {"trades": 8, "sharpe": 2, "sortino": 3, "profit_factor": 0.9, "max_drawdown": -0.05},
            min_trades=12,
        )
        self.assertFalse(weak["pass"])
        self.assertIn("oos_trades<12", weak["reason"])
        self.assertIn("oos_profit_factor<1.15", weak["reason"])

        robust = holdout_acceptance(
            {"trades": 30, "sharpe": 1.5, "sortino": 2, "profit_factor": 1.8, "max_drawdown": -0.08},
            min_trades=12,
        )
        self.assertTrue(robust["pass"])

    def test_meta_gate_blocks_2y_oos_sample_below_30(self):
        close = pd.Series([100 + index * 0.2 for index in range(40)], dtype="float64")
        frame = pd.DataFrame(
            {
                "open": close - 0.1,
                "high": close + 0.5,
                "low": close - 0.5,
                "close": close,
                "volume_usd": 1_000_000.0,
            }
        )
        row = {
            "strategy": "2",
            "ticker": "TEST",
            "window": "2y",
            "trades": 60,
            "min_trades": 30,
            "sharpe": 1.5,
            "sortino": 2.0,
            "profit_factor": 1.8,
            "win_rate": 0.55,
            "avg_trade_roi": 0.01,
            "max_drawdown": -0.08,
            "walk_forward_pass": True,
            "oos_pass": True,
            "oos_trades": 18,
        }
        decision = meta_alpha_decision(row, 1, frame)
        self.assertIn("oos_sample_below_support_minimum", decision["blocked_reasons"])
        self.assertFalse(decision["opener_allowed"])


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