import json from dataclasses import replace from pathlib import Path import numpy as np import pandas as pd import pytest from psx_signal.config import Settings, UniverseSelectionConfig from psx_signal.paper.study import ( ProspectiveSignal, StudyStore, deterministic_random_order, entry_stress, session_exit_date, ) from psx_signal.pipelines.prospective_paper import ( ProspectivePaperService, simulate_prospective_portfolio, supported_current_membership, study_configuration, ) def _signal(sequence: int = 1, symbol: str = "AAA", memberships=("TOP_1", "TOP_3")): return ProspectiveSignal.create( sequence=sequence, strategy_id="psx_xgb_5d_topn_v1", strategy_fingerprint="a8cf67171c3e6d2e478b923d071bc5c281ccbf7afe7cd4f7129ff7029378d733", model_version="m1", dataset_version="d1", feature_version="f1", as_of_date="2026-08-19", symbol=symbol, rank=sequence, buy_score=.7, market_regime="NEUTRAL", signal_day_close=100.0, planned_entry_session="NEXT_PSX_SESSION", planned_exit_session="T_PLUS_5_CLOSE", holding_sessions=5, membership_source="official.pdf", membership_status="VERIFIED_MEMBER", data_completeness_status="COMPLETE", equity_provider="yahoo_finance", benchmark_provider="karandaaz", raw_artifacts=("raw.parquet",), selection_memberships=memberships, ) def test_prospective_date_cannot_precede_study_start(tmp_path: Path) -> None: service = ProspectivePaperService(Settings(), tmp_path, start_date="2026-08-19") with pytest.raises(ValueError, match="cannot precede"): service.eod(as_of="2026-08-18") def test_duplicate_signal_cannot_overwrite_immutable_original(tmp_path: Path) -> None: store = StudyStore(tmp_path) store.initialize(study_configuration("2026-08-19")) signal = _signal() store.append_signals([signal]) original = store.signals_path.read_text() with pytest.raises(ValueError, match="Prospective signal"): store.append_signals([_signal()]) assert store.signals_path.read_text() == original def test_append_only_journal_is_idempotent_by_event_key(tmp_path: Path) -> None: store = StudyStore(tmp_path); store.initialize(study_configuration("2026-08-19")) assert store.append_event("EOD_RESULT", event_key="day-1", status="NO_ELIGIBLE_SIGNAL") assert not store.append_event("EOD_RESULT", event_key="day-1", status="GENERATED") assert len(store.events()) == 1 def test_no_signal_day_is_recorded_without_lowering_threshold(tmp_path: Path, monkeypatch) -> None: service = ProspectivePaperService(Settings(), tmp_path, start_date="2026-08-19") rows = pd.DataFrame({"symbol": ["AAA"], "f": [1.0], "close": [100.0], "market_regime": ["NEUTRAL"]}) members = pd.DataFrame({"symbol": ["AAA"], "source": ["PSX"]}) monkeypatch.setattr(service, "_data_gate", lambda *args: { "status": "COMPLETE", "reason": "", "as_of_date": "2026-08-19", "equity_rows": 1, "benchmark_available": True, "universe_size": 1, "verified_member_count": 1, "complete_feature_rows": 1, "error_count": 0, "warning_count": 0, "feature_rows": rows, "members": members, "model_metadata": {"feature_columns": ["f"]}, }) monkeypatch.setattr(service, "_record_drift", lambda *args: None) monkeypatch.setattr(service, "_record_random_control", lambda *args: None) monkeypatch.setattr("psx_signal.pipelines.prospective_paper.ProbabilisticModel.load", lambda *args: type("M", (), {"predict_proba": lambda self, x: np.array([[.4, .3, .3]])})()) result = service.eod(as_of="2026-08-19") assert result["status"] == "NO_ELIGIBLE_SIGNAL" assert service.store.processed_session("psx_xgb_5d_topn_v1", "2026-08-19")["status"] == "NO_ELIGIBLE_SIGNAL" def test_incomplete_benchmark_records_skipped_session(tmp_path: Path) -> None: data_root = tmp_path / "data" equity_path = data_root / "normalized/equities/eod.parquet" equity_path.parent.mkdir(parents=True) pd.DataFrame({"date": pd.to_datetime(["2026-08-19"]), "symbol": ["AAA"], "open": [1.0], "high": [1.0], "low": [1.0], "close": [1.0], "volume": [1.0]}).to_parquet(equity_path, index=False) settings = replace(Settings(), data=replace(Settings().data, root=str(data_root))) service = ProspectivePaperService(settings, tmp_path / "study", start_date="2026-08-19") result = service.eod(as_of="2026-08-19") assert result["status"] == "SKIPPED_DATA_INCOMPLETE" def test_membership_unknown_fails_safely(tmp_path: Path) -> None: data_root = tmp_path / "data" dates = pd.to_datetime(["2026-08-18", "2026-08-19"]) equities = pd.DataFrame({"date": dates, "symbol": ["AAA", "AAA"], "open": [1.0, 1.0], "high": [1.0, 1.0], "low": [1.0, 1.0], "close": [1.0, 1.0], "volume": [100.0, 100.0]}) equity_path = data_root / "normalized/equities/eod.parquet"; equity_path.parent.mkdir(parents=True) equities.to_parquet(equity_path, index=False) benchmark_path = data_root / "normalized/indices/KSE100.parquet"; benchmark_path.parent.mkdir(parents=True) pd.DataFrame({"date": dates, "close": [1.0, 1.0]}).to_parquet(benchmark_path, index=False) selection = UniverseSelectionConfig(size=1, minimum_history_days=1, minimum_trading_frequency=0, liquidity_lookback_days=1) settings = replace(Settings(), data=replace(Settings().data, root=str(data_root)), universe_selection=selection, minimum_history=1) result = ProspectivePaperService(settings, tmp_path / "study").eod(as_of="2026-08-19") assert result["status"] == "SKIPPED_DATA_INCOMPLETE" assert result["reason"] == "Official membership UNKNOWN" def test_latest_official_snapshot_expires_at_next_scheduled_recomposition(tmp_path: Path) -> None: path = tmp_path / "normalized/kse100_constituent_snapshots.parquet"; path.parent.mkdir(parents=True) pd.DataFrame({"effective_from": pd.to_datetime(["2026-04-01"]), "symbol": ["AAA"], "index": ["KSE100"], "source_reference": ["official.pdf"]}).to_parquet(path, index=False) assert supported_current_membership(tmp_path, pd.Timestamp("2026-08-19"))["symbol"].tolist() == ["AAA"] assert supported_current_membership(tmp_path, pd.Timestamp("2026-10-01")).empty def test_session_counting_entry_stress_and_random_control_are_deterministic() -> None: sessions = pd.to_datetime(["2026-01-05", "2026-01-06", "2026-01-07", "2026-01-08", "2026-01-09"]) assert session_exit_date(sessions, "2026-01-05", 5) == pd.Timestamp("2026-01-09") stress = entry_stress(100.0, 110.0) assert stress["entry_open_plus_0.0100"] == pytest.approx(101.0) assert stress["gross_return_at_0.0100"] == pytest.approx(110 / 101 - 1) assert deterministic_random_order(["B", "A", "C"], "2026-01-01") == deterministic_random_order(["B", "A", "C"], "2026-01-01") def test_top1_and_top3_daily_sleeves_equal_weight_and_apply_fees() -> None: signals = [json.loads(json.dumps(_signal(i, symbol, ("TOP_3",)).__dict__, default=list)) for i, symbol in enumerate(("AAA", "BBB", "CCC"), start=1)] entries = {s["signal_id"]: {"entry_date": "2026-01-05", "observed_entry_open": 100.0} for s in signals} exits = {s["signal_id"]: {"exit_date": "2026-01-09", "observed_exit_close": 110.0} for s in signals} sessions = pd.date_range("2026-01-05", periods=5, freq="D") equities = pd.DataFrame([{"date": day, "symbol": symbol, "close": 105.0} for day in sessions for symbol in ("AAA", "BBB", "CCC")]) trades, ledger = simulate_prospective_portfolio( signals, entries, exits, set(), equities, sessions, "TOP_3", True, ) allocations = [trade["allocated_capital"] for trade in trades] assert allocations == pytest.approx([allocations[0]] * 3) assert sum(allocations) == pytest.approx(200_000.0) assert all(trade["entry_fee"] > 0 and trade["exit_fee"] > 0 for trade in trades) assert ledger[-1]["open_positions"] == 0 def test_corporate_action_flag_is_append_only_and_update_is_idempotent(tmp_path: Path) -> None: service = ProspectivePaperService(Settings(), tmp_path, start_date="2026-08-19") service.store.append_signals([_signal()]) assert service.mark_corporate_action(_signal().signal_id, "2026-08-20", "INCOMPARABLE") assert not service.mark_corporate_action(_signal().signal_id, "2026-08-20", "INCOMPARABLE") first = service.update(); second = service.update() assert first["status"] == second["status"] == "NO_CHANGES" def test_review_rules_and_t1_open_t5_close_update(tmp_path: Path) -> None: data_root = tmp_path / "data" sessions = pd.to_datetime(["2026-01-02", "2026-01-05", "2026-01-06", "2026-01-07", "2026-01-08", "2026-01-09"]) equities = pd.DataFrame({"date": sessions, "symbol": ["AAA"] * len(sessions), "open": [99, 101, 102, 103, 104, 105], "high": [100, 102, 103, 104, 105, 111], "low": [98, 100, 101, 102, 103, 104], "close": [100, 101, 102, 103, 104, 110], "volume": [1000] * len(sessions)}) equity_path = data_root / "normalized/equities/eod.parquet"; equity_path.parent.mkdir(parents=True) equities.to_parquet(equity_path, index=False) benchmark_path = data_root / "normalized/indices/KSE100.parquet"; benchmark_path.parent.mkdir(parents=True) pd.DataFrame({"date": sessions, "close": range(len(sessions))}).to_parquet(benchmark_path, index=False) settings = replace(Settings(), data=replace(Settings().data, root=str(data_root))) service = ProspectivePaperService(settings, tmp_path / "study", start_date="2026-01-02") signal = _signal(); payload = signal.__dict__.copy(); payload["as_of_date"] = "2026-01-02" signal = ProspectiveSignal(**payload) service.store.append_signals([signal]) with pytest.raises(ValueError, match="objective"): service.review("2026-01-02", "PREMARKET_REVIEWED", "DISLIKE_COMPANY") service.review("2026-01-02", "PREMARKET_REVIEWED", "VERIFIED_TRADABLE") result = service.update() assert result == {"status": "UPDATED", "entries": 1, "exits": 1} events = service.store.events() entry = next(row for row in events if row["event_type"] == "ENTRY") exit_event = next(row for row in events if row["event_type"] == "EXIT") assert entry["entry_date"] == "2026-01-05" assert entry["observed_entry_open"] == 101 assert exit_event["exit_date"] == "2026-01-09" assert exit_event["observed_exit_close"] == 110 assert service.update()["status"] == "NO_CHANGES"