test_v3_p0_integrity.py

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

import numpy as np
import pandas as pd

from v3_continuity_binding import (
    annotations_for_event,
    build_binding,
    event_is_post_repair_formal_eligible,
)
from v3_build_identity import identity_files
from v3_forward_event_engine import validate_bar_contract
from v3_paper_account import account_risk_snapshot, empty_account_state, evaluate_position_exit
from v3_state_transaction import commit_text_bundle
from v3_underlying_intraday_collector import aggregate_completed_bars, fetch_yahoo_chart


class V3P0IntegrityTests(unittest.TestCase):
    def test_source_identity_tracks_runtime_not_dashboard_presentation(self):
        paths = {path.name for path in identity_files()}
        self.assertIn("v3_shadow_runtime.py", paths)
        self.assertIn("v3_paper_account.py", paths)
        self.assertNotIn("product_dashboard.py", paths)

    def test_numpy_boolean_is_preserved_as_formal_eligibility(self):
        contract = validate_bar_contract({
            "open": 100.0, "high": 101.0, "low": 99.0, "close": 100.5,
            "bar_close_time": "2026-07-23T14:00:00Z",
            "available_time": "2026-07-23T14:00:01Z",
            "formal_realtime_evidence_eligible": np.bool_(True),
        }, "2026-07-23T14:00:02Z", "15m")
        self.assertTrue(contract["formal_realtime_evidence_eligible"])

    def test_recovered_minutes_cannot_create_formal_completed_bar(self):
        opens = pd.date_range("2026-07-23T13:30:00Z", periods=15, freq="1min")
        minutes = pd.DataFrame({
            "bar_open_time": opens,
            "bar_close_time": opens + pd.Timedelta(minutes=1),
            "first_observed_at": opens + pd.Timedelta(minutes=11),
            "open": 100.0,
            "high": 101.0,
            "low": 99.0,
            "close": 100.5,
            "volume": 10.0,
            "market_phase": "regular",
            "capture_class": "sleep_gap_recovered",
            "formal_realtime_evidence_eligible": False,
        })
        bars = aggregate_completed_bars(
            minutes, "15m", observed_at="2026-07-23T14:00:00Z",
        )
        self.assertEqual(len(bars), 1)
        self.assertFalse(bool(bars.iloc[0]["formal_realtime_evidence_eligible"]))
        self.assertFalse(bool(bars.iloc[0]["formal_forward_eligible"]))
        self.assertEqual(bars.iloc[0]["capture_class"], "sleep_gap_recovered")

    def test_exit_engine_excludes_any_bar_that_started_before_entry(self):
        entry = pd.Timestamp("2026-07-23T14:05:00Z")
        position = {
            "entry_time": entry.isoformat(),
            "entry_price": 100.0,
            "direction": "LONG",
            "timeframe": "15m",
            "risk_plan": {
                "stop_loss": 96.0,
                "take_profit": 110.0,
                "max_holding_days": 5,
            },
        }
        frame = pd.DataFrame({
            "bar_open_time": [
                "2026-07-23T14:00:00Z", "2026-07-23T14:15:00Z",
            ],
            "bar_close_time": [
                "2026-07-23T14:15:00Z", "2026-07-23T14:30:00Z",
            ],
            "available_time": [
                "2026-07-23T14:15:01Z", "2026-07-23T14:30:01Z",
            ],
            "open": [100.0, 95.0], "high": [101.0, 96.0],
            "low": [90.0, 94.0], "close": [95.0, 95.0],
        })
        exit_result = evaluate_position_exit(position, frame, "2026-07-23T14:30:01Z")
        self.assertEqual(exit_result["exit_time"], "2026-07-23T14:30:00+00:00")
        self.assertEqual(exit_result["exit_reason"], "stop_loss")
        self.assertTrue(exit_result["entry_bar_excluded"])

    def test_mark_to_market_drawdown_blocks_new_entries(self):
        state = empty_account_state(run_id="p0")
        state["risk_state"] = {
            "trading_date": "2026-07-23",
            "high_water_equity_usd": 100_000.0,
            "daily_start_equity_usd": 100_000.0,
        }
        state["open_positions"] = [{
            "position_id": "p1", "ticker": "IBM", "timeframe": "15m",
            "entry_time": "2026-07-23T14:00:00Z", "entry_price": 100.0,
            "direction": "LONG", "notional_usd": 100_000.0,
            "round_trip_cost_rate": 0.0008,
        }]
        frame = pd.DataFrame({
            "bar_close_time": ["2026-07-23T14:15:00Z"],
            "available_time": ["2026-07-23T14:15:01Z"],
            "close": [80.0],
        })
        risk = account_risk_snapshot(
            state, frames={("IBM", "15m"): frame},
            as_of="2026-07-23T14:20:00Z",
        )
        self.assertFalse(risk["new_entries_allowed"])
        self.assertIn("account_drawdown_kill_switch", risk["entry_blockers"])

    def test_bundle_commit_updates_ledger_and_account_together(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            ledger = root / "ledger.jsonl"
            account = root / "account.json"
            result = commit_text_bundle({
                ledger: json.dumps({"event_id": "event-1"}) + "\n",
                account: json.dumps({"paper_trading_only": True}),
            }, transaction_root=root / "transactions", transaction_kind="test")
            self.assertEqual(result["status"], "RECOVERED_AND_COMMITTED")
            self.assertIn("event-1", ledger.read_text(encoding="utf-8"))
            self.assertTrue(json.loads(account.read_text(encoding="utf-8"))["paper_trading_only"])

    def test_post_repair_identity_does_not_reclassify_legacy_event(self):
        binding = build_binding(
            "run-p0", effective_from="2026-07-23T14:00:00Z",
        )
        legacy = {
            "run_id": "run-p0", "decision_time": "2026-07-23T13:59:59Z",
            "formal_realtime_evidence_eligible": True,
            "data_freshness_verified": True,
        }
        post_repair = {
            "run_id": "run-p0", "decision_time": "2026-07-23T14:00:00Z",
            "formal_realtime_evidence_eligible": True,
            "data_freshness_verified": True,
        }
        legacy.update(annotations_for_event(legacy, binding=binding))
        post_repair.update(annotations_for_event(post_repair, binding=binding))
        self.assertFalse(event_is_post_repair_formal_eligible(legacy, binding=binding))
        self.assertTrue(event_is_post_repair_formal_eligible(post_repair, binding=binding))

    def test_underlying_fetch_has_process_level_network_deadline(self):
        payload = {"chart": {"result": [], "error": None}}
        completed = SimpleNamespace(returncode=0, stdout=json.dumps(payload), stderr="")
        with mock.patch("v3_underlying_intraday_collector.subprocess.run", return_value=completed) as run:
            self.assertEqual(fetch_yahoo_chart("AAPL", timeout=8), payload)
        command = run.call_args.args[0]
        self.assertIn("--connect-timeout", command)
        self.assertIn("--max-time", command)
        self.assertEqual(run.call_args.kwargs["timeout"], 10)


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