test_v3_logic_repair.py

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

import pandas as pd

import dashboard_v3
import v3_system_coordinator
from us_equity_factor_library import apply_us_equity_factor_filter
from v3_clawby_router import load_capabilities
from v3_execution_engine import build_risk_plan
from v3_portfolio_engine import build_virtual_subaccounts, construct_portfolio
from v3_shadow_pipeline import process_shadow_pipeline
from v3_factor_context import build_crypto_factor_context


class V3LogicRepairTests(unittest.TestCase):
    def _candidate(self, ticker="IBM", direction="LONG", timeframe="15m", score=85.0, robot_id="G1-US01"):
        return {
            "qualified": True, "run_id": "run-v3", "robot_id": robot_id, "group_id": robot_id[:2],
            "market": "US_EQUITY", "ticker": ticker, "sector": "Technology",
            "timeframe": timeframe, "signal_bar_close": "2026-07-15T14:00:00+00:00",
            "decision_time": "2026-07-15T14:00:02+00:00", "direction": direction,
            "final_score": score, "reliability": 0.8, "annualized_volatility": 0.25,
            "blockers": [], "paper_trading_only": True,
        }

    def _frame(self):
        close = pd.Series([100 + index * 0.05 for index in range(40)], dtype=float)
        frame = pd.DataFrame({
            "open": close - 0.02, "high": close + 0.1, "low": close - 0.1,
            "close": close, "volume": 1000.0,
            "bar_close_time": pd.date_range("2026-07-15T04:15:00Z", periods=40, freq="15min"),
        })
        frame["available_time"] = frame["bar_close_time"] + pd.Timedelta(seconds=1)
        return frame

    @staticmethod
    def _execution_quote(decision_time, price=102.0):
        decision = pd.Timestamp(decision_time)
        request_started = decision + pd.Timedelta(milliseconds=100)
        provider_time = decision + pd.Timedelta(milliseconds=500)
        received = decision + pd.Timedelta(seconds=1)
        return {
            "quote_time": received.isoformat(),
            "quote_price": price,
            "quote_source": "synthetic_bitget_snapshot",
            "directional_price_side": "ask",
            "instrument_validation": {
                "passed": True, "is_tradable_for_paper": True,
                "execution_symbol": "IBMUSDT",
                "basis_reference_eligible": True, "spread_pct": 0.001,
                "basis_gap_pct": 0.002, "issues": [],
            },
            "execution_provenance": {
                "provider_quote_time": provider_time.isoformat(),
                "request_started_at": request_started.isoformat(),
                "received_at": received.isoformat(),
                "execution_available_time": received.isoformat(),
                "time_provenance_verified": True,
            },
        }

    def test_current_clawby_limit_is_1200_with_official_account_verification(self):
        capabilities = load_capabilities()
        self.assertEqual(capabilities["rate_limit_per_minute"], 1200)
        self.assertEqual(capabilities["rate_limit_verification_status"], "OFFICIAL_ACCOUNT_VERIFIED")

    def test_daily_formula_cannot_treat_short_bars_as_days(self):
        frame = self._frame()
        signal = pd.Series(1, index=frame.index)
        self.assertIsNone(apply_us_equity_factor_filter(
            signal, frame, "multi_horizon_time_series_trend", timeframe="15m"
        ))

    def test_fresh_wrapper_cannot_hide_stale_crypto_signal_bar(self):
        context = build_crypto_factor_context(
            ticker="BTC", direction="LONG", decision_time="2026-07-15T09:00:00Z",
            factor_ids=["ema_trend"], asset={},
            signal={"generated_at": "2026-07-15T08:59:59Z", "last_bar_at": "2026-07-13T12:00:00Z", "direction": "LONG"},
        )
        self.assertFalse(context["ema_trend"]["passed"])
        self.assertFalse(context["ema_trend"]["source_fresh"])

    def test_crypto_local_pit_factors_use_real_completed_data(self):
        close = pd.Series([100 + index for index in range(40)], dtype=float)
        frame = pd.DataFrame({
            "bar_close_time": pd.date_range("2026-07-13T18:00:00Z", periods=40, freq="1h"),
            "available_time": pd.date_range("2026-07-13T18:00:05Z", periods=40, freq="1h"),
            "open": close - 0.5, "high": close + 1, "low": close - 1, "close": close,
            "volume": 1000, "funding_rate": [-0.001] * 39 + [-0.01],
            "oi_value": [1000 + index * 10 for index in range(40)],
        })
        decision = frame.iloc[-1]["available_time"]
        context = build_crypto_factor_context(
            ticker="BTC", direction="LONG", decision_time=decision,
            factor_ids=["ema_trend", "oi_change", "realized_vol_guard", "cvd_flow"],
            asset={}, signal={}, frame=frame,
        )
        self.assertTrue(context["ema_trend"]["passed"])
        self.assertTrue(context["oi_change"]["passed"])
        self.assertTrue(context["realized_vol_guard"]["pit_valid"])
        self.assertEqual(context["cvd_flow"]["status"], "UNAVAILABLE")

    def test_cross_timeframe_opposite_directions_conflict_at_instrument_level(self):
        left = self._candidate(direction="LONG", timeframe="15m", score=85, robot_id="G1-US01")
        right = self._candidate(direction="SHORT", timeframe="1h", score=83, robot_id="G2-US01")
        right["signal_bar_close"] = "2026-07-15T14:30:00+00:00"
        portfolio = construct_portfolio(build_virtual_subaccounts([left, right]))
        self.assertEqual(portfolio["portfolio_trade_count"], 0)
        self.assertTrue(any(row["status"] == "CONFLICT_NO_TRADE" for row in portfolio["conflicts"]))

    def test_missing_correlation_does_not_default_to_zero(self):
        left = self._candidate(ticker="IBM", robot_id="G1-US01")
        right = self._candidate(ticker="QCOM", robot_id="G2-US01")
        portfolio = construct_portfolio(build_virtual_subaccounts([left, right]), correlation={})
        self.assertEqual(portfolio["portfolio_trade_count"], 1)
        self.assertTrue(any(row["status"] == "CORRELATION_INPUT_MISSING_NO_TRADE" for row in portfolio["conflicts"]))

    def test_frozen_risk_plan_uses_existing_parameters(self):
        long_plan = build_risk_plan(100.0, "LONG")
        short_plan = build_risk_plan(100.0, "SHORT")
        self.assertEqual((long_plan["stop_loss"], long_plan["take_profit"]), (96.0, 108.0))
        self.assertEqual((short_plan["stop_loss"], short_plan["take_profit"]), (104.0, 92.0))
        self.assertEqual(long_plan["max_holding_days"], 5)

    def _write_status_fixture(self, root: Path, shadow_status: str):
        result_dir = root / "results"
        result_dir.mkdir()
        shared = root / "shared"
        shared.mkdir()
        configs = root / "configs"
        configs.mkdir()
        files = {
            "runtime.json": {
                "paper_trading_only": True, "automatic_promotion_allowed": False,
                "legacy_evidence_policy": "read_only_archive_not_active_evidence",
                "run_id": "r", "architecture_version": "v3", "mode": "shadow",
                "formal_forward_collection_active": False,
            },
            "timeframes.json": {"paper_trading_only": True, "completed_bars_only": True, "same_bar_execution_forbidden": True},
            "sources.json": {
                "paper_trading_only": True, "rate_limit_per_minute": 1200,
                "rate_limit_source": "Clawby official account API with user-confirmed local ceiling",
                "rate_limit_verification_status": "USER_CONFIRMED_PENDING_OFFICIAL_ACCOUNT_REFRESH",
                "sources": {"clawby": {"may_create_signal": False, "may_hard_veto_paper_entry": False}},
            },
            "factors.json": {"automatic_activation_allowed": False},
            "validation.json": {"status": "WAIT_FOR_FROZEN_V3_CALIBRATOR", "records": {}},
            "decision.json": {"score_scale_maximum": 100, "minimum_alert_score": 70},
        }
        for name, payload in files.items():
            (configs / name).write_text(json.dumps(payload), encoding="utf-8")
        (shared / "clawby_rate_limit_status.json").write_text(json.dumps({"local_limit": 1200}), encoding="utf-8")
        (result_dir / "clawby_rate_limit_status.json").write_text(json.dumps({"local_limit": 1200}), encoding="utf-8")
        now = pd.Timestamp.now(tz="UTC").isoformat()
        (result_dir / "v3_underlying_intraday_status.json").write_text(json.dumps({"status": "HEALTHY"}), encoding="utf-8")
        (result_dir / "v3_intraday_capture_service_status.json").write_text(json.dumps({
            "status": "HEALTHY",
            "generated_at": now,
            "capture_interval_seconds": 60,
            "timely_capture_max_delay_seconds": 180,
            "interval_within_timely_gate": True,
        }), encoding="utf-8")
        (result_dir / "v3_signal_funnel_status.json").write_text(json.dumps({
            "schema_version": "v3_signal_funnel_1", "stages": [{"stage": "raw_signal", "count": 0}],
        }), encoding="utf-8")
        (result_dir / "v3_shadow_runtime_status.json").write_text(json.dumps({
            "generated_at": now, "status": shadow_status,
            "robots": {"raw_signals": 0, "crypto_raw_signals": 0, "institutional_raw_signals": 0},
            "shadow_forward_events": [], "qualified_alerts": [], "virtual_positions": [],
            "shadow_event_ledger_path": str(root / "data/v3_shadow/robot_event_ledger.jsonl"),
            "paper_account_path": str(root / "data/v3_paper_next/paper_account.json"),
            "independent_forward": {}, "account_risk_before_cycle": {}, "robot_subaccounts": {},
            "runtime_capabilities": {
                "clawby_api_key_available": True,
                "clawby_rate_ledger_local_present": True,
                "crypto_completed_frames_operational": True,
                "execution_time_provenance_enforced": True,
                "structured_cost_model_version": "test-cost-model",
                "shadow_pipeline_error_free": True,
                "role_quorum_pipeline_error_free": True,
                "continuity_binding": {
                    "status": "ACTIVE_VALID",
                    "evidence_epoch_id": "fixture_epoch",
                    "source_identity_sha256": "fixture_sha",
                    "formal_entry_allowed": True,
                },
            },
            "shadow_pipeline_errors": [],
            "role_quorum_pipeline_errors": [],
            "formal_forward_ledgers_modified": False,
        }), encoding="utf-8")
        return result_dir, shared, configs

    def test_system_never_calls_empty_validation_or_degraded_runtime_ready(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            result_dir, shared, configs = self._write_status_fixture(root, "HEALTHY")
            patches = (
                mock.patch.object(v3_system_coordinator, "RESULTS_DIR", result_dir),
                mock.patch.object(v3_system_coordinator, "SHARED_RESULTS_DIR", shared),
                mock.patch.object(v3_system_coordinator, "RUNTIME_PATH", configs / "runtime.json"),
                mock.patch.object(v3_system_coordinator, "TIMEFRAME_PATH", configs / "timeframes.json"),
                mock.patch.object(v3_system_coordinator, "DATA_SOURCE_PATH", configs / "sources.json"),
                mock.patch.object(v3_system_coordinator, "FACTOR_OVERLAY_PATH", configs / "factors.json"),
                mock.patch.object(v3_system_coordinator, "VALIDATION_PATH", configs / "validation.json"),
                mock.patch.object(v3_system_coordinator, "DECISION_POLICY_PATH", configs / "decision.json"),
                mock.patch.object(
                    v3_system_coordinator,
                    "strategy_control_snapshot",
                    return_value={
                        "control_health": "VALID",
                        "controls_effective": True,
                        "fail_closed": False,
                    },
                ),
                mock.patch.object(
                    v3_system_coordinator,
                    "runtime_binding_status",
                    return_value={
                        "status": "ACTIVE_VALID",
                        "evidence_epoch_id": "fixture_epoch",
                        "source_identity_sha256": "fixture_sha",
                        "formal_entry_allowed": True,
                    },
                ),
            )
            with patches[0], patches[1], patches[2], patches[3], patches[4], patches[5], patches[6], patches[7], patches[8], patches[9]:
                status = v3_system_coordinator.build_status()
                self.assertEqual(
                    status["status"],
                    "WAIT_FOR_FROZEN_V3_CALIBRATOR",
                    {key: value for key, value in status["checks"].items() if value is not True},
                )
                shadow = json.loads((result_dir / "v3_shadow_runtime_status.json").read_text())
                shadow["status"] = "HEALTHY_WITH_ISOLATED_INPUTS"
                (result_dir / "v3_shadow_runtime_status.json").write_text(json.dumps(shadow))
                self.assertEqual(v3_system_coordinator.build_status()["status"], "WAIT_FOR_FROZEN_V3_CALIBRATOR")
                shadow["runtime_capabilities"]["continuity_binding"]["evidence_epoch_id"] = "stale_epoch"
                (result_dir / "v3_shadow_runtime_status.json").write_text(json.dumps(shadow))
                self.assertEqual(v3_system_coordinator.build_status()["status"], "FIX_INFRASTRUCTURE")
                shadow["runtime_capabilities"]["continuity_binding"]["evidence_epoch_id"] = "fixture_epoch"
                shadow["runtime_capabilities"]["crypto_completed_frames_operational"] = False
                (result_dir / "v3_shadow_runtime_status.json").write_text(json.dumps(shadow))
                self.assertEqual(v3_system_coordinator.build_status()["status"], "FIX_INFRASTRUCTURE")
                shadow["runtime_capabilities"]["crypto_completed_frames_operational"] = True
                shadow["status"] = "DEGRADED"
                (result_dir / "v3_shadow_runtime_status.json").write_text(json.dumps(shadow))
                self.assertEqual(v3_system_coordinator.build_status()["status"], "FIX_INFRASTRUCTURE")

    def test_dashboard_does_not_promote_preview_to_alert(self):
        shadow = {
            "status": "HEALTHY", "robots": {}, "qualified_alerts": [],
            "qualified_signal_preview": [{"ticker": "IBM"}],
        }
        def fake_read(path, default):
            return shadow if path.name == "v3_shadow_runtime_status.json" else default
        with mock.patch.object(dashboard_v3, "build_system_status", return_value={"checks": {}}), \
             mock.patch.object(dashboard_v3, "build_registry_status", return_value={"robots": []}), \
             mock.patch.object(dashboard_v3, "read_json", side_effect=fake_read), \
             mock.patch.object(dashboard_v3, "ledger_rows", return_value=[]):
            payload = dashboard_v3.build_payload()
        self.assertEqual(payload["realtime"]["qualified_alerts"], [])
        self.assertEqual(len(payload["realtime"]["qualified_signal_preview"]), 1)

    def test_end_to_end_shadow_event_is_captured_alerted_and_idempotent(self):
        frame = self._frame()
        candidate = self._candidate()
        candidate["signal_bar_close"] = frame.iloc[-1]["bar_close_time"].isoformat()
        candidate["decision_time"] = frame.iloc[-1]["available_time"].isoformat()
        portfolio = construct_portfolio(build_virtual_subaccounts([candidate]))
        quote = self._execution_quote(candidate["decision_time"])
        quote_time = quote["quote_time"]
        with tempfile.TemporaryDirectory() as directory, \
             mock.patch("v3_shadow_pipeline.load_validated_execution_quote", return_value=quote):
            ledger = Path(directory) / "shadow.jsonl"
            account = Path(directory) / "account.json"
            kwargs = {
                "evaluations": [candidate], "portfolio": portfolio,
                "frames": {("IBM", "15m"): frame}, "as_of": quote_time,
                "shared_root": Path(directory), "ledger_path": ledger,
                "account_path": account,
            }
            first = process_shadow_pipeline(**kwargs)
            second = process_shadow_pipeline(**kwargs)
            self.assertEqual(len(first["events"]), 1)
            self.assertEqual(first["events"][0]["execution_status"], "captured")
            self.assertEqual(len(first["alerts"]), 1)
            self.assertEqual(len(ledger.read_text().splitlines()), 1)
            self.assertEqual(second["events"][0]["event_id"], first["events"][0]["event_id"])

    def test_executed_event_matures_after_signal_disappears(self):
        frame = self._frame()
        candidate = self._candidate()
        candidate["signal_bar_close"] = frame.iloc[-1]["bar_close_time"].isoformat()
        candidate["decision_time"] = frame.iloc[-1]["available_time"].isoformat()
        portfolio = construct_portfolio(build_virtual_subaccounts([candidate]))
        quote = self._execution_quote(candidate["decision_time"])
        quote_time = quote["quote_time"]
        later = frame.copy()
        extension_times = pd.date_range(frame.iloc[-1]["bar_close_time"] + pd.Timedelta(minutes=15), periods=8, freq="15min")
        extension = pd.DataFrame({
            "open": 102.0, "high": 103.0, "low": 101.5, "close": 102.5, "volume": 1200.0,
            "bar_close_time": extension_times,
        })
        extension["available_time"] = extension["bar_close_time"] + pd.Timedelta(seconds=1)
        later = pd.concat([later, extension], ignore_index=True)
        with tempfile.TemporaryDirectory() as directory, \
             mock.patch("v3_shadow_pipeline.load_validated_execution_quote", return_value=quote):
            ledger = Path(directory) / "shadow.jsonl"
            account = Path(directory) / "account.json"
            process_shadow_pipeline(
                evaluations=[candidate], portfolio=portfolio, frames={("IBM", "15m"): frame},
                as_of=quote_time, shared_root=Path(directory), ledger_path=ledger,
                account_path=account, active_run_id="run-v3",
            )
            matured = process_shadow_pipeline(
                evaluations=[], portfolio={"trades": []}, frames={("IBM", "15m"): later},
                as_of=extension.iloc[-1]["available_time"], shared_root=Path(directory), ledger_path=ledger,
                account_path=account, active_run_id="run-v3",
            )
        self.assertEqual(len(matured["events"]), 1)
        self.assertGreaterEqual(matured["events"][0].get("mature_horizon_count", 0), 1)

    def test_deduplicated_trade_alert_uses_primary_robot(self):
        frame = self._frame()
        primary = self._candidate(score=88.0, robot_id="G1-US01")
        peer = self._candidate(score=82.0, robot_id="G5-US07")
        for candidate in (primary, peer):
            candidate["signal_bar_close"] = frame.iloc[-1]["bar_close_time"].isoformat()
            candidate["decision_time"] = frame.iloc[-1]["available_time"].isoformat()
        portfolio = construct_portfolio(build_virtual_subaccounts([primary, peer]))
        quote = self._execution_quote(primary["decision_time"])
        quote_time = quote["quote_time"]
        with tempfile.TemporaryDirectory() as directory, \
             mock.patch("v3_shadow_pipeline.load_validated_execution_quote", return_value=quote):
            result = process_shadow_pipeline(
                evaluations=[primary, peer], portfolio=portfolio,
                frames={("IBM", "15m"): frame}, as_of=quote_time,
                shared_root=Path(directory), ledger_path=Path(directory) / "events.jsonl",
                account_path=Path(directory) / "account.json",
            )
        self.assertEqual(len(result["alerts"]), 1)
        self.assertEqual(result["alerts"][0]["robot_id"], "G1-US01")
        self.assertEqual(result["virtual_positions"][0]["robot_id"], "G1-US01")


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