from __future__ import annotations import json from dataclasses import replace from datetime import UTC, date, datetime from pathlib import Path import numpy as np import pandas as pd from psx_signal.config import Settings from psx_signal.data.providers.pypsx_provider import ( PyPsxProviderUnavailable, PyPsxToolkitProvider, ) from psx_signal.data.storage import DatasetStore from psx_signal.pipelines.daily_operations import DailyBenchmarkRefreshService from psx_signal.pipelines.prospective_paper import ProspectivePaperService def _settings(tmp_path: Path) -> Settings: return replace(Settings(), data=replace(Settings().data, root=str(tmp_path / "data"))) def _history(include_active: bool = False, conflict: bool = False) -> pd.DataFrame: dates = list(pd.bdate_range("2026-07-01", "2026-08-13")) dates += [pd.Timestamp("2026-08-17"), pd.Timestamp("2026-08-18")] if include_active: dates.append(pd.Timestamp("2026-08-19")) close = 180_000 + np.arange(len(dates)) * 100.0 if conflict: close[5] *= 1.02 return pd.DataFrame({ "Date": dates, "Open": close - 20, "High": close + 50, "Low": close - 50, "Close": close, "Volume": np.arange(len(dates)) + 1000, }).set_index("Date") def _anchor(settings: Settings) -> pd.DataFrame: source = _history().reset_index() source = source[source["Date"] <= pd.Timestamp("2026-08-13")] frame = pd.DataFrame({ "index": "KSE100", "date": source["Date"], "open": pd.NA, "high": pd.NA, "low": pd.NA, "close": source["Close"], "volume": pd.NA, "provider": "karandaaz", "provider_trust": "RESEARCH_SECONDARY", "source_reference": "existing immutable Karandaaz anchor", }) store = DatasetStore(settings.data.root) DatasetStore._atomic_parquet(frame, store.kse100_path) return frame class FakeTicker: calls: list[tuple[str, str, bool]] = [] frame = _history() def __init__(self, symbol: str) -> None: assert symbol == "KSE100" def get_historical(self, start_date: str, end_date: str, show_progress: bool = False): self.calls.append((start_date, end_date, show_progress)) return self.frame.copy() class FakeToolkit: Ticker = FakeTicker def test_pypsx_retrieval_uses_documented_kse100_historical_interface() -> None: FakeTicker.calls.clear() result = PyPsxToolkitProvider(FakeToolkit).get_historical( date(2026, 7, 1), date(2026, 8, 18), ) assert result.status == "SUCCESS" assert FakeTicker.calls == [("2026-07-01", "2026-08-18", False)] def test_automatic_aug17_18_backfill_transition_and_provenance(tmp_path: Path) -> None: settings = _settings(tmp_path); _anchor(settings) service = DailyBenchmarkRefreshService( settings, tmp_path / "study", PyPsxToolkitProvider(FakeToolkit), now=datetime(2026, 8, 19, 10, tzinfo=UTC), ) result = service.run() assert result.status == "SUCCESS" and result.rows_added == 2 assert result.latest == "2026-08-18" assert result.transition_status == "VALIDATED_RESEARCH_TRANSITION" transition = json.loads((tmp_path / "study/benchmark_provider_transition.json").read_text()) assert transition["matching_dates"] >= 10 assert transition["median_relative_difference"] == 0 frame = DatasetStore(settings.data.root).read(DatasetStore(settings.data.root).kse100_path) added = frame[frame["provider"] == "pypsx_toolkit"] assert added["date"].dt.date.tolist() == [date(2026, 8, 17), date(2026, 8, 18)] assert set(added["provider_trust"]) == {"RESEARCH_SECONDARY"} assert set(added["underlying_market"]) == {"PSX"} assert added["raw_artifact"].notna().all() assert service.run().status == "NO_NEW_DATA" assert len(DatasetStore(settings.data.root).read(DatasetStore(settings.data.root).kse100_path)) == len(frame) def test_transition_conflict_blocks_benchmark_append(tmp_path: Path) -> None: settings = _settings(tmp_path); anchor = _anchor(settings) FakeTicker.frame = _history(conflict=True) try: result = DailyBenchmarkRefreshService( settings, tmp_path / "study", PyPsxToolkitProvider(FakeToolkit), now=datetime(2026, 8, 19, 10, tzinfo=UTC), ).run() finally: FakeTicker.frame = _history() assert result.status == "PROVIDER_TRANSITION_CONFLICT" frame = DatasetStore(settings.data.root).read(DatasetStore(settings.data.root).kse100_path) assert len(frame) == len(anchor) def test_existing_benchmark_conflict_is_rejected_without_overwrite(tmp_path: Path) -> None: settings = _settings(tmp_path); anchor = _anchor(settings) conflicting = pd.DataFrame([{ "index": "KSE100", "date": pd.Timestamp("2026-08-17"), "close": 999_999.0, "provider": "pypsx_toolkit", "provider_trust": "RESEARCH_SECONDARY", }]) store = DatasetStore(settings.data.root) DatasetStore._atomic_parquet(pd.concat([anchor, conflicting], ignore_index=True), store.kse100_path) result = DailyBenchmarkRefreshService( settings, tmp_path / "study", PyPsxToolkitProvider(FakeToolkit), now=datetime(2026, 8, 19, 10, tzinfo=UTC), ).run() assert result.status == "BENCHMARK_DATA_CONFLICT" frame = store.read(store.kse100_path) original = frame[frame["date"] == pd.Timestamp("2026-08-17")].iloc[0] assert original["close"] == 999_999.0 def test_active_session_is_not_appended(tmp_path: Path) -> None: settings = _settings(tmp_path); _anchor(settings) FakeTicker.frame = _history(include_active=True) try: result = DailyBenchmarkRefreshService( settings, tmp_path / "study", PyPsxToolkitProvider(FakeToolkit), now=datetime(2026, 8, 19, 10, tzinfo=UTC), ).run() finally: FakeTicker.frame = _history() assert result.latest == "2026-08-18" frame = DatasetStore(settings.data.root).read(DatasetStore(settings.data.root).kse100_path) assert date(2026, 8, 19) not in set(frame["date"].dt.date) def test_pypsx_outage_returns_waiting_status(tmp_path: Path) -> None: settings = _settings(tmp_path); _anchor(settings) class Outage: def get_historical(self, start, end): raise PyPsxProviderUnavailable("provider unavailable") result = DailyBenchmarkRefreshService( settings, tmp_path / "study", Outage(), now=datetime(2026, 8, 19, 10, tzinfo=UTC), ).run() assert result.status == "WAITING_FOR_BENCHMARK_PROVIDER" assert "provider unavailable" in result.error def test_paper_signal_records_pypsx_benchmark_provenance(tmp_path: Path, monkeypatch) -> None: settings = _settings(tmp_path); store = DatasetStore(settings.data.root) equities = pd.DataFrame({ "date": pd.to_datetime(["2026-08-19"]), "symbol": ["AAA"], "open": [100.0], "high": [101.0], "low": [99.0], "close": [100.0], "volume": [1000], "provider": ["yahoo_finance"], }) DatasetStore._atomic_parquet(equities, store.equities_path) manifest = store.root / "normalized/manifest.json"; manifest.parent.mkdir(parents=True, exist_ok=True) manifest.write_text(json.dumps({ "dataset_version": "d1", "benchmark_provider": "pypsx_toolkit", }), encoding="utf-8") service = ProspectivePaperService(settings, tmp_path / "study", start_date="2026-08-19") 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": pd.DataFrame({ "symbol": ["AAA"], "f": [1.0], "close": [100.0], "market_regime": ["NEUTRAL"], "provider": ["yahoo_finance"], }), "members": pd.DataFrame({"symbol": ["AAA"], "source_reference": ["official.pdf"]}), "model_metadata": {"feature_columns": ["f"], "model_version": "m1", "feature_version": "f1"}, }) 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([[.1, .2, .7]])})()) result = service.eod(as_of="2026-08-19") assert result["status"] == "GENERATED" assert service.store.signals()[0]["benchmark_provider"] == "pypsx_toolkit"