from __future__ import annotations import argparse import json import os from dataclasses import asdict from datetime import date from pathlib import Path import pandas as pd from psx_signal.config import load_settings from psx_signal.data.comparison import compare_sources, read_manual_observations from psx_signal.data.providers import CsvDataProvider from psx_signal.data.validation import DataValidator from psx_signal.data.storage import DatasetStore from psx_signal.data.universe import UniverseSelector from psx_signal.pipelines.data_sync import DataSyncService, data_status from psx_signal.pipelines.yahoo_bootstrap import YahooBootstrapService from psx_signal.pipelines.benchmark_import import BenchmarkImportService from psx_signal.pipelines.dataset import create_feature_dataset, import_csv, write_parquet from psx_signal.pipelines.predict import generate_eod_predictions from psx_signal.pipelines.training import train_model from psx_signal.pipelines.walk_forward import walk_forward_backtest from psx_signal.experiments.baseline_v1 import run_readiness_gate from psx_signal.experiments.task6_runner import run_task6 from psx_signal.pipelines.paper_signal import format_paper_signals, generate_paper_signals from psx_signal.pipelines.prospective_paper import ( OPERATIONAL_REVIEW_REASONS, ProspectivePaperService, format_eod_result, ) from psx_signal.pipelines.daily_operations import ( BenchmarkAppendService, DailyBenchmarkRefreshService, DailyEquityRefreshService, OfficialDailyIngestService, ProspectiveDailyService, ) def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="psx-signal", description="PSX signal research engine") parser.add_argument("--config", default=os.getenv("PSX_SIGNAL_CONFIG", "config/default.toml")) subparsers = parser.add_subparsers(dest="command", required=True) ingest = subparsers.add_parser("ingest", help="Copy a CSV into immutable raw storage") ingest.add_argument("--input", required=True) ingest.add_argument("--raw-dir", default="data/raw") validate = subparsers.add_parser("validate-data", help="Validate raw or normalized OHLCV") validate.add_argument("--input") validate.add_argument("--benchmark") validate.add_argument("--symbol") validate.add_argument("--report", default="reports/data-quality.json") backfill_data = subparsers.add_parser("backfill", help="Backfill official-file or licensed PSX data") backfill_data.add_argument("--start", required=True, type=date.fromisoformat) backfill_data.add_argument("--end", required=True, type=date.fromisoformat) sync_data = subparsers.add_parser("sync-data", help="Import only available/missing PSX EOD artifacts") sync_data.add_argument("--date", type=date.fromisoformat) subparsers.add_parser("data-status", help="Show actual normalized dataset coverage and quality") bootstrap_yahoo = subparsers.add_parser( "bootstrap-yahoo", help="Bootstrap research-only PSX history from Yahoo Finance" ) bootstrap_yahoo.add_argument("--start", type=date.fromisoformat, default=date(2016, 1, 1)) bootstrap_yahoo.add_argument("--end", type=date.fromisoformat, default=date.today()) bootstrap_yahoo.add_argument("--symbols-file", default="data/reference/psx_symbols.csv") bootstrap_yahoo.add_argument("--limit", type=int) bootstrap_yahoo.add_argument("--force", action="store_true") compare = subparsers.add_parser( "compare-provider", help="Compare manual authoritative PSX observations with normalized data" ) compare.add_argument("--reference", choices=["psx"], required=True) compare.add_argument("--candidate", choices=["yahoo"], required=True) compare.add_argument( "--reference-file", default="data/reference/psx_manual_observations.csv" ) compare.add_argument("--output", default="reports/task4/provider_comparison.csv") import_benchmark = subparsers.add_parser( "import-benchmark", help="Import a manually downloaded research benchmark CSV" ) import_benchmark.add_argument( "--input", default="data/inbox/benchmark/kse100_karandaaz.csv" ) universe = subparsers.add_parser("select-universe", help="Select liquid universe using point-in-time data") universe.add_argument("--as-of", required=True) universe.add_argument("--output", default="reports/initial-universe.csv") export_data = subparsers.add_parser("export-data", help="Export normalized Parquet dataset to CSV") export_data.add_argument( "--dataset", choices=["equities", "kse100", "stock-master", "corporate-actions", "constituents"], required=True, ) export_data.add_argument("--output", required=True) experiment = subparsers.add_parser("experiment", help="Run a reproducible research experiment") experiment.add_argument("name", choices=["baseline-v1", "task6-robustness"]) experiment.add_argument("--output-dir") paper = subparsers.add_parser("paper-signal", help="Append frozen research paper signals") paper.add_argument("--strategy", required=True) paper.add_argument("--as-of", default="latest") paper.add_argument("--ledger", default="reports/paper/signals.jsonl") paper_eod = subparsers.add_parser("paper-eod", help="Run prospective EOD shadow-paper gate") paper_eod.add_argument("--strategy", required=True) paper_eod.add_argument("--study-root", default="reports/paper/study_v1") paper_review = subparsers.add_parser("paper-review", help="Record objective pre-market review") paper_review.add_argument("--as-of", required=True) paper_review.add_argument("--status", required=True, choices=[ "PREMARKET_REVIEWED", "CANCELLED_NOT_TRADABLE", "CANCELLED_DATA_ERROR", "CANCELLED_CORPORATE_ACTION", "CANCELLED_MARKET_EVENT", ]) paper_review.add_argument("--reason", required=True, choices=sorted(OPERATIONAL_REVIEW_REASONS)) paper_review.add_argument("--study-root", default="reports/paper/study_v1") paper_update = subparsers.add_parser("paper-update", help="Record observed paper entries and exits") paper_update.add_argument("--study-root", default="reports/paper/study_v1") paper_status = subparsers.add_parser("paper-status", help="Show prospective study status") paper_status.add_argument("--study-root", default="reports/paper/study_v1") subparsers.add_parser( "daily-refresh-equities", help="Append finalized Yahoo PSX equity sessions only" ) benchmark_refresh = subparsers.add_parser( "daily-refresh-benchmark", help="Append completed pyPSX Toolkit KSE100 sessions" ) benchmark_refresh.add_argument("--study-root", default="reports/paper/study_v1") subparsers.add_parser( "ingest-psx-daily", help="Ingest manually downloaded official PSX daily artifacts" ) benchmark_add = subparsers.add_parser( "benchmark-add", help="Append an immutable official KSE100 close observation" ) benchmark_add.add_argument("--index", required=True) benchmark_add.add_argument("--date", required=True, type=date.fromisoformat) benchmark_add.add_argument("--close", required=True, type=float) benchmark_add.add_argument("--source", required=True) benchmark_add.add_argument("--source-reference", required=True) prospective_daily = subparsers.add_parser( "prospective-daily", help="Run the ordered prospective daily operations workflow" ) prospective_daily.add_argument("--study-root", default="reports/paper/study_v1") features = subparsers.add_parser("build-features", help="Build leakage-safe feature dataset") features.add_argument("--input", required=True) features.add_argument("--benchmark") features.add_argument("--output", default="data/features/features.parquet") features.add_argument("--horizons", default="1,5") features.add_argument("--universe", help="Comma-separated symbols; overrides config universe") train = subparsers.add_parser("train", help="Train and persist a model") train.add_argument("--features", default="data/features/features.parquet") train.add_argument("--model", choices=["logistic", "xgboost"], default="logistic") train.add_argument("--horizon", type=int, choices=[1, 5, 10], default=1) train.add_argument("--output") backtest = subparsers.add_parser("backtest", help="Run expanding-window out-of-sample backtest") backtest.add_argument("--features", default="data/features/features.parquet") backtest.add_argument("--model", choices=["majority", "momentum", "logistic", "xgboost"], default="logistic") backtest.add_argument("--horizon", type=int, choices=[1, 5, 10], default=1) backtest.add_argument("--report") for name in ("predict", "eod"): predict = subparsers.add_parser(name, help="Generate append-only EOD predictions") predict.add_argument("--features", default="data/features/features.parquet") predict.add_argument("--artifact", required=True) predict.add_argument("--as-of") predict.add_argument("--symbol", action="append", dest="symbols") predict.add_argument("--ledger", default="reports/predictions.jsonl") return parser def main(argv: list[str] | None = None) -> int: args = _parser().parse_args(argv) settings = load_settings(args.config) if args.command == "ingest": print(import_csv(args.input, args.raw_dir)) return 0 if args.command == "validate-data": input_path = Path(args.input) if args.input else Path(settings.data.root) / "normalized" / "equities" / "eod.parquet" if not input_path.exists(): print(f"No dataset found at {input_path}. Add official PSX artifacts and run psx-signal sync-data.") return 1 if input_path.suffix == ".parquet": frame = pd.read_parquet(input_path) benchmark_path = Path(args.benchmark) if args.benchmark else Path(settings.data.root) / "normalized" / "indices" / "KSE100.parquet" benchmark_dates = pd.read_parquet(benchmark_path)["date"] if benchmark_path.exists() else None else: provider = CsvDataProvider(input_path, args.benchmark) frame = provider.get_ohlcv() try: benchmark_dates = provider.get_index_history(settings.benchmark_symbol)["date"] except ValueError: benchmark_dates = None if args.symbol: frame = frame[frame["symbol"] == args.symbol.upper()] validator = DataValidator( settings.extreme_return_threshold, settings.validation.get("expected_session_coverage", 0.90), settings.minimum_history, ) report = validator.report(validator.validate(frame, benchmark_dates), args.report) print(json.dumps(report, indent=2)) return 0 if report["valid"] else 2 if args.command in {"backfill", "sync-data"}: service = DataSyncService(settings) result = service.backfill(args.start, args.end) if args.command == "backfill" else service.sync(args.date) print(json.dumps(asdict(result), indent=2, default=str)) return 0 if result.status == "SUCCESS" else 1 if result.status in {"PARTIAL", "NO_DATA"} else 2 if args.command == "bootstrap-yahoo": result = YahooBootstrapService(settings).run( args.start, args.end, args.symbols_file, args.limit, args.force ) print(json.dumps(asdict(result), indent=2, default=str)) return 0 if result.status == "SUCCESS" else 1 if result.status == "PARTIAL" else 2 if args.command == "import-benchmark": result = BenchmarkImportService(settings).run(args.input) print(json.dumps(asdict(result), indent=2, default=str)) return 0 if result.status == "SUCCESS" else 2 if args.command == "compare-provider": reference = read_manual_observations(args.reference_file) candidate = DatasetStore(settings.data.root).read( DatasetStore(settings.data.root).equities_path ) candidate = candidate[candidate.get("provider", "") == "yahoo_finance"] compared = compare_sources(reference, candidate) Path(args.output).parent.mkdir(parents=True, exist_ok=True) compared.to_csv(args.output, index=False) counts = compared["comparison"].value_counts().to_dict() if not compared.empty else {} print(json.dumps({"records_checked": len(compared), **counts, "output": args.output}, indent=2)) return 0 if not compared.empty else 1 if args.command == "data-status": status = data_status(settings) if status.get("status") == "NO_DATA": print(status["message"]) return 1 latest = status.get("latest_sync", {}) print( "PSX DATA STATUS\n===============\n\n" f"Equities\nStart: {status.get('equity_start_date')}\n" f"End: {status.get('equity_end_date')}\n" f"Trading sessions: {status.get('trading_sessions')}\n" f"Symbols observed: {status.get('symbols')}\nRows: {status.get('rows')}\n\n" f"KSE100\nStart: {status.get('kse100_start_date')}\n" f"End: {status.get('kse100_end_date')}\nRows: {status.get('kse100_rows')}\n\n" f"Quality\nErrors: {status.get('validation_errors')}\n" f"Warnings: {status.get('validation_warnings')}\nInfo: {status.get('validation_info')}\n\n" f"Corporate Actions\nKnown events: {status.get('corporate_actions')}\n" f"Unresolved gaps: {status.get('unresolved_discontinuities')}\n\n" f"Latest sync\nStatus: {latest.get('status')}\n" f"Timestamp: {latest.get('timestamp')}\nProvider: {latest.get('provider')}\n" ) return 0 if args.command == "select-universe": store = DatasetStore(settings.data.root) equities = store.read(store.equities_path) indices = store.read(store.kse100_path) master = store.read(store.stock_master_path) selected = UniverseSelector(settings.universe_selection).select( equities, args.as_of, indices["date"] if not indices.empty else None, master ) Path(args.output).parent.mkdir(parents=True, exist_ok=True) selected.to_csv(args.output, index=False) print(selected.to_string(index=False)) return 0 if args.command == "export-data": store = DatasetStore(settings.data.root) paths = { "equities": store.equities_path, "kse100": store.kse100_path, "stock-master": store.stock_master_path, "corporate-actions": store.corporate_actions_path, "constituents": store.constituents_path, } frame = store.read(paths[args.dataset]) if frame.empty: raise ValueError(f"Dataset {args.dataset!r} is empty or unavailable") Path(args.output).parent.mkdir(parents=True, exist_ok=True) frame.to_csv(args.output, index=False) print(args.output) return 0 if args.command == "experiment": if args.name == "task6-robustness": result = run_task6(settings, args.output_dir or "reports/task6") print(json.dumps(result, indent=2)) return 0 result = run_readiness_gate(settings, args.output_dir or "reports/task3") print(json.dumps(asdict(result), indent=2)) return 0 if result.status in {"READY", "READY_WITH_LIMITATIONS"} else 2 if args.command == "paper-signal": records = generate_paper_signals( settings, args.strategy, as_of=args.as_of, ledger_path=args.ledger, ) print(format_paper_signals(records)) return 0 if args.command == "paper-eod": service = ProspectivePaperService(settings, args.study_root) print(format_eod_result(service.eod(args.strategy))) return 0 if args.command == "paper-review": result = ProspectivePaperService(settings, args.study_root).review( args.as_of, args.status, args.reason, ) print(json.dumps(result, indent=2)) return 0 if args.command == "paper-update": result = ProspectivePaperService(settings, args.study_root).update() print(json.dumps(result, indent=2)) return 0 if args.command == "paper-status": result = ProspectivePaperService(settings, args.study_root).status() print( "PROSPECTIVE STUDY\n=================\n\n" f"Study start: {result['study_start_date']}\n" f"Completed sessions: {result['completed_sessions']}\n" "Target: 60 minimum / 120 preferred\n" f"Trading dates processed: {result['trading_dates_processed']}\n" f"Top-1 generated: {result['top1_generated']}\n" f"Top-1 entered: {result['top1_entered']}\n" f"Top-1 completed: {result['top1_completed']}\n" f"No-signal sessions: {result['no_signal_sessions']}\n" f"Operational cancellations: {result['operational_cancellations']}\n" f"Data-incomplete sessions: {result['data_incomplete_sessions']}\n" f"Open positions: {result['open_positions']}\n" f"Next gate: {result['next_checkpoint']}" ) return 0 if args.command == "daily-refresh-equities": result = DailyEquityRefreshService(settings).run() print(json.dumps(asdict(result), indent=2, default=str)) return 0 if result.status in {"SUCCESS", "UP_TO_DATE"} else 1 if args.command == "daily-refresh-benchmark": result = DailyBenchmarkRefreshService(settings, args.study_root).run() print(json.dumps(asdict(result), indent=2, default=str)) return 0 if result.status in {"SUCCESS", "NO_NEW_DATA"} else 1 if args.command == "ingest-psx-daily": result = OfficialDailyIngestService(settings).run() print(json.dumps(asdict(result), indent=2, default=str)) return 0 if result.status in {"SUCCESS", "NO_NEW_FILES"} else 1 if args.command == "benchmark-add": result = BenchmarkAppendService(settings).add( args.index, args.date, args.close, args.source, args.source_reference, ) print(json.dumps(result, indent=2, default=str)) return 0 if result["status"] in {"ADDED", "NO_CHANGE"} else 2 if args.command == "prospective-daily": result = ProspectiveDailyService(settings, args.study_root).run() print(json.dumps(result, indent=2, default=str)) return 0 if result["status"] in {"COMPLETE", "LATE_EOD_GENERATION"} else 1 if args.command == "build-features": horizons = tuple(int(item.strip()) for item in args.horizons.split(",")) unsupported = set(horizons) - set(settings.horizons.values()) if unsupported: raise ValueError(f"Unconfigured horizons: {sorted(unsupported)}") universe = tuple(item.strip().upper() for item in args.universe.split(",")) if args.universe else None result = create_feature_dataset(args.input, args.benchmark, settings, horizons, universe) print(write_parquet(result, args.output)) return 0 if args.command == "train": output = args.output or f"models/{args.model}_{args.horizon}d.joblib" metadata = train_model(pd.read_parquet(args.features), settings, args.model, args.horizon, output) print(json.dumps(metadata, indent=2)) return 0 if args.command == "backtest": report_path = args.report or f"reports/{args.model}_{args.horizon}d_backtest.json" report = walk_forward_backtest( pd.read_parquet(args.features), settings, args.model, args.horizon, report_path ) print(json.dumps(report, indent=2)) return 0 if args.command in {"predict", "eod"}: records = generate_eod_predictions( pd.read_parquet(args.features), settings, args.artifact, as_of=args.as_of, symbols=args.symbols, ledger_path=args.ledger, ) print(json.dumps([asdict(record) for record in records], indent=2)) return 0 return 1 if __name__ == "__main__": raise SystemExit(main())