diff --git a/qt/cli.py b/qt/cli.py index 63751e8..ca5c81d 100644 --- a/qt/cli.py +++ b/qt/cli.py @@ -206,6 +206,7 @@ def _cmd_run_phase_i5d_intraday_groups(args: argparse.Namespace) -> int: def _cmd_run_eval_jump_amount_corr(args: argparse.Namespace) -> int: """Run the two real jump-amount-corr factor evaluations (cache-only) + reports.""" from qt.eval_jump_amount_corr import run_eval_jump_amount_corr + from qt.exec_basis_eval import format_exec_basis_line try: result = run_eval_jump_amount_corr(args.config) @@ -225,12 +226,14 @@ def _cmd_run_eval_jump_amount_corr(args: argparse.Namespace) -> int: f"dashboards: {result.reports.no_book_dashboard} | " f"{result.reports.with_book_dashboard}" ) + print(format_exec_basis_line(result.exec_basis)) return 0 def _cmd_run_eval_minute_ideal_amplitude(args: argparse.Namespace) -> int: """Run the two real minute-ideal-amplitude factor evaluations (cache-only) + reports.""" from qt.eval_minute_ideal_amplitude import run_eval_minute_ideal_amplitude + from qt.exec_basis_eval import format_exec_basis_line try: result = run_eval_minute_ideal_amplitude(args.config) @@ -250,12 +253,14 @@ def _cmd_run_eval_minute_ideal_amplitude(args: argparse.Namespace) -> int: f"dashboards: {result.reports.no_book_dashboard} | " f"{result.reports.with_book_dashboard}" ) + print(format_exec_basis_line(result.exec_basis)) return 0 def _cmd_run_eval_amp_marginal_anomaly_vol(args: argparse.Namespace) -> int: """Run the two real amp-marginal-anomaly-vol factor evaluations (cache-only) + reports.""" from qt.eval_amp_marginal_anomaly_vol import run_eval_amp_marginal_anomaly_vol + from qt.exec_basis_eval import format_exec_basis_line try: result = run_eval_amp_marginal_anomaly_vol(args.config) @@ -275,12 +280,14 @@ def _cmd_run_eval_amp_marginal_anomaly_vol(args: argparse.Namespace) -> int: f"dashboards: {result.reports.no_book_dashboard} | " f"{result.reports.with_book_dashboard}" ) + print(format_exec_basis_line(result.exec_basis)) return 0 def _cmd_run_eval_volume_peak_count(args: argparse.Namespace) -> int: """Run the two real volume-peak-count factor evaluations (cache-only) + reports.""" from qt.eval_volume_peak_count import run_eval_volume_peak_count + from qt.exec_basis_eval import format_exec_basis_line try: result = run_eval_volume_peak_count(args.config) @@ -300,12 +307,14 @@ def _cmd_run_eval_volume_peak_count(args: argparse.Namespace) -> int: f"dashboards: {result.reports.no_book_dashboard} | " f"{result.reports.with_book_dashboard}" ) + print(format_exec_basis_line(result.exec_basis)) return 0 def _cmd_run_eval_intraday_amp_cut(args: argparse.Namespace) -> int: """Run the two real intraday-amp-cut factor evaluations (cache-only) + reports.""" from qt.eval_intraday_amp_cut import run_eval_intraday_amp_cut + from qt.exec_basis_eval import format_exec_basis_line try: result = run_eval_intraday_amp_cut(args.config) @@ -325,12 +334,14 @@ def _cmd_run_eval_intraday_amp_cut(args: argparse.Namespace) -> int: f"dashboards: {result.reports.no_book_dashboard} | " f"{result.reports.with_book_dashboard}" ) + print(format_exec_basis_line(result.exec_basis)) return 0 def _cmd_run_eval_peak_interval_kurtosis(args: argparse.Namespace) -> int: """Run the two real peak-interval-kurtosis evaluations (cache-only) + reports.""" from qt.eval_peak_interval_kurtosis import run_eval_peak_interval_kurtosis + from qt.exec_basis_eval import format_exec_basis_line try: result = run_eval_peak_interval_kurtosis(args.config) @@ -350,12 +361,14 @@ def _cmd_run_eval_peak_interval_kurtosis(args: argparse.Namespace) -> int: f"dashboards: {result.reports.no_book_dashboard} | " f"{result.reports.with_book_dashboard}" ) + print(format_exec_basis_line(result.exec_basis)) return 0 def _cmd_run_eval_valley_relative_vwap(args: argparse.Namespace) -> int: """Run the two real valley-relative-VWAP evaluations (cache-only) + reports.""" from qt.eval_valley_relative_vwap import run_eval_valley_relative_vwap + from qt.exec_basis_eval import format_exec_basis_line try: result = run_eval_valley_relative_vwap(args.config) @@ -375,6 +388,7 @@ def _cmd_run_eval_valley_relative_vwap(args: argparse.Namespace) -> int: f"dashboards: {result.reports.no_book_dashboard} | " f"{result.reports.with_book_dashboard}" ) + print(format_exec_basis_line(result.exec_basis)) return 0 @@ -398,6 +412,7 @@ def _fmt_metric(value: object, spec: str = ".4f") -> str: def _cmd_run_eval_ridge_minute_return(args: argparse.Namespace) -> int: """Run the two real ridge-minute-return evaluations (cache-only) + reports.""" from qt.eval_ridge_minute_return import run_eval_ridge_minute_return + from qt.exec_basis_eval import format_exec_basis_line try: result = run_eval_ridge_minute_return(args.config) @@ -433,12 +448,14 @@ def _cmd_run_eval_ridge_minute_return(args: argparse.Namespace) -> int: f"dashboards: {result.reports.no_book_dashboard} | " f"{result.reports.with_book_dashboard}" ) + print(format_exec_basis_line(result.exec_basis)) return 0 def _cmd_run_eval_peak_ridge_amount_ratio(args: argparse.Namespace) -> int: """Run the two real peak/ridge amount-ratio evaluations (cache-only) + reports.""" from qt.eval_peak_ridge_amount_ratio import run_eval_peak_ridge_amount_ratio + from qt.exec_basis_eval import format_exec_basis_line try: result = run_eval_peak_ridge_amount_ratio(args.config) @@ -477,12 +494,14 @@ def _cmd_run_eval_peak_ridge_amount_ratio(args: argparse.Namespace) -> int: f"dashboards: {result.reports.no_book_dashboard} | " f"{result.reports.with_book_dashboard}" ) + print(format_exec_basis_line(result.exec_basis)) return 0 def _cmd_run_eval_valley_price_quantile(args: argparse.Namespace) -> int: """Run the two real valley-price-quantile evaluations (cache-only) + reports.""" from qt.eval_valley_price_quantile import run_eval_valley_price_quantile + from qt.exec_basis_eval import format_exec_basis_line try: result = run_eval_valley_price_quantile(args.config) @@ -527,12 +546,14 @@ def _cmd_run_eval_valley_price_quantile(args: argparse.Namespace) -> int: f"dashboards: {result.reports.no_book_dashboard} | " f"{result.reports.with_book_dashboard}" ) + print(format_exec_basis_line(result.exec_basis)) return 0 def _cmd_run_eval_valley_ridge_vwap_ratio(args: argparse.Namespace) -> int: """Run the two real valley/ridge VWAP-ratio evaluations (cache-only) + reports.""" from qt.eval_valley_ridge_vwap_ratio import run_eval_valley_ridge_vwap_ratio + from qt.exec_basis_eval import format_exec_basis_line try: result = run_eval_valley_ridge_vwap_ratio(args.config) @@ -555,6 +576,7 @@ def _cmd_run_eval_valley_ridge_vwap_ratio(args: argparse.Namespace) -> int: f"dashboards: {result.reports.no_book_dashboard} | " f"{result.reports.with_book_dashboard}" ) + print(format_exec_basis_line(result.exec_basis)) return 0 diff --git a/qt/eval_amp_marginal_anomaly_vol.py b/qt/eval_amp_marginal_anomaly_vol.py index 5ebd269..0914789 100644 --- a/qt/eval_amp_marginal_anomaly_vol.py +++ b/qt/eval_amp_marginal_anomaly_vol.py @@ -59,6 +59,7 @@ from factors.compute.intraday_derived import AmpMarginalAnomalyVolFactor from factors.spec import FactorSpec from qt.config import RootConfig, load_config +from qt.exec_basis_eval import ExecBasisEvaluation, run_exec_basis_evaluation from qt.pipeline import ( _build_cache, _build_universe, @@ -360,6 +361,7 @@ class AmpAnomalyVolEvalResult: no_book_metrics: dict with_book_metrics: dict reports: _RunReports + exec_basis: ExecBasisEvaluation log_path: Path elapsed: float @@ -452,6 +454,22 @@ def run_eval_amp_marginal_anomaly_vol(config_path: str) -> AmpAnomalyVolEvalResu report_dir=Path(cfg.output.report_dir), ) + # Second evaluation basis (task card §2.3): the SAME factor values scored on + # the 14:51 VWAP exec-to-exec return instead of close-to-close. The reports + # above are the close_to_close control and are left untouched. + exec_basis = run_exec_basis_evaluation( + factor_series, + spec, + eval_cfg, + book_processed, + cfg=cfg, + panel=panel, + symbols=symbols, + logger=logger, + report_dir=Path(cfg.output.report_dir), + stem=_REPORT_STEM, + ) + no_book_metrics = extract_metrics(reports.no_book) with_book_metrics = extract_metrics(reports.with_book) logger.info( @@ -472,6 +490,7 @@ def run_eval_amp_marginal_anomaly_vol(config_path: str) -> AmpAnomalyVolEvalResu no_book_metrics=no_book_metrics, with_book_metrics=with_book_metrics, reports=reports, + exec_basis=exec_basis, log_path=log_path, elapsed=time.monotonic() - started, ) diff --git a/qt/eval_intraday_amp_cut.py b/qt/eval_intraday_amp_cut.py index 42b13cf..eb12140 100644 --- a/qt/eval_intraday_amp_cut.py +++ b/qt/eval_intraday_amp_cut.py @@ -70,6 +70,7 @@ from factors.compute.intraday_derived import IntradayAmpCutFactor from factors.spec import FactorSpec from qt.config import RootConfig, load_config +from qt.exec_basis_eval import ExecBasisEvaluation, run_exec_basis_evaluation from qt.pipeline import ( _build_cache, _build_universe, @@ -381,6 +382,7 @@ class AmpCutEvalResult: no_book_metrics: dict with_book_metrics: dict reports: _RunReports + exec_basis: ExecBasisEvaluation log_path: Path elapsed: float @@ -474,6 +476,22 @@ def run_eval_intraday_amp_cut(config_path: str) -> AmpCutEvalResult: report_dir=Path(cfg.output.report_dir), ) + # Second evaluation basis (task card §2.3): the SAME factor values scored on + # the 14:51 VWAP exec-to-exec return instead of close-to-close. The reports + # above are the close_to_close control and are left untouched. + exec_basis = run_exec_basis_evaluation( + factor_series, + spec, + eval_cfg, + book_processed, + cfg=cfg, + panel=panel, + symbols=symbols, + logger=logger, + report_dir=Path(cfg.output.report_dir), + stem=_REPORT_STEM, + ) + no_book_metrics = extract_metrics(reports.no_book) with_book_metrics = extract_metrics(reports.with_book) logger.info( @@ -495,6 +513,7 @@ def run_eval_intraday_amp_cut(config_path: str) -> AmpCutEvalResult: no_book_metrics=no_book_metrics, with_book_metrics=with_book_metrics, reports=reports, + exec_basis=exec_basis, log_path=log_path, elapsed=time.monotonic() - started, ) diff --git a/qt/eval_jump_amount_corr.py b/qt/eval_jump_amount_corr.py index 267d071..c2f0850 100644 --- a/qt/eval_jump_amount_corr.py +++ b/qt/eval_jump_amount_corr.py @@ -53,6 +53,7 @@ from factors.compute.intraday_derived import JumpAmountCorrFactor from factors.spec import FactorSpec from qt.config import RootConfig, load_config +from qt.exec_basis_eval import ExecBasisEvaluation, run_exec_basis_evaluation from qt.pipeline import ( _build_cache, _build_universe, @@ -346,6 +347,7 @@ class JumpEvalResult: no_book_metrics: dict with_book_metrics: dict reports: _RunReports + exec_basis: ExecBasisEvaluation log_path: Path elapsed: float @@ -433,6 +435,22 @@ def run_eval_jump_amount_corr(config_path: str) -> JumpEvalResult: report_dir=Path(cfg.output.report_dir), ) + # Second evaluation basis (task card §2.3): the SAME factor values scored on + # the 14:51 VWAP exec-to-exec return instead of close-to-close. The reports + # above are the close_to_close control and are left untouched. + exec_basis = run_exec_basis_evaluation( + factor_series, + spec, + eval_cfg, + book_processed, + cfg=cfg, + panel=panel, + symbols=symbols, + logger=logger, + report_dir=Path(cfg.output.report_dir), + stem=_REPORT_STEM, + ) + no_book_metrics = extract_metrics(reports.no_book) with_book_metrics = extract_metrics(reports.with_book) logger.info( @@ -453,6 +471,7 @@ def run_eval_jump_amount_corr(config_path: str) -> JumpEvalResult: no_book_metrics=no_book_metrics, with_book_metrics=with_book_metrics, reports=reports, + exec_basis=exec_basis, log_path=log_path, elapsed=time.monotonic() - started, ) diff --git a/qt/eval_minute_ideal_amplitude.py b/qt/eval_minute_ideal_amplitude.py index 682d813..9759b4a 100644 --- a/qt/eval_minute_ideal_amplitude.py +++ b/qt/eval_minute_ideal_amplitude.py @@ -56,6 +56,7 @@ from factors.compute.intraday_derived import MinuteIdealAmplitudeFactor from factors.spec import FactorSpec from qt.config import RootConfig, load_config +from qt.exec_basis_eval import ExecBasisEvaluation, run_exec_basis_evaluation from qt.pipeline import ( _build_cache, _build_universe, @@ -354,6 +355,7 @@ class MinuteIdealAmpEvalResult: no_book_metrics: dict with_book_metrics: dict reports: _RunReports + exec_basis: ExecBasisEvaluation log_path: Path elapsed: float @@ -445,6 +447,22 @@ def run_eval_minute_ideal_amplitude(config_path: str) -> MinuteIdealAmpEvalResul report_dir=Path(cfg.output.report_dir), ) + # Second evaluation basis (task card §2.3): the SAME factor values scored on + # the 14:51 VWAP exec-to-exec return instead of close-to-close. The reports + # above are the close_to_close control and are left untouched. + exec_basis = run_exec_basis_evaluation( + factor_series, + spec, + eval_cfg, + book_processed, + cfg=cfg, + panel=panel, + symbols=symbols, + logger=logger, + report_dir=Path(cfg.output.report_dir), + stem=_REPORT_STEM, + ) + no_book_metrics = extract_metrics(reports.no_book) with_book_metrics = extract_metrics(reports.with_book) logger.info( @@ -465,6 +483,7 @@ def run_eval_minute_ideal_amplitude(config_path: str) -> MinuteIdealAmpEvalResul no_book_metrics=no_book_metrics, with_book_metrics=with_book_metrics, reports=reports, + exec_basis=exec_basis, log_path=log_path, elapsed=time.monotonic() - started, ) diff --git a/qt/eval_peak_interval_kurtosis.py b/qt/eval_peak_interval_kurtosis.py index 70e62d6..b21a08f 100644 --- a/qt/eval_peak_interval_kurtosis.py +++ b/qt/eval_peak_interval_kurtosis.py @@ -68,6 +68,7 @@ from factors.compute.intraday_derived import PeakIntervalKurtosisFactor from factors.spec import FactorSpec from qt.config import RootConfig, load_config +from qt.exec_basis_eval import ExecBasisEvaluation, run_exec_basis_evaluation from qt.pipeline import ( _build_cache, _build_universe, @@ -376,6 +377,7 @@ class PeakIntervalKurtosisEvalResult: no_book_metrics: dict with_book_metrics: dict reports: _RunReports + exec_basis: ExecBasisEvaluation log_path: Path elapsed: float @@ -471,6 +473,22 @@ def run_eval_peak_interval_kurtosis(config_path: str) -> PeakIntervalKurtosisEva report_dir=Path(cfg.output.report_dir), ) + # Second evaluation basis (task card §2.3): the SAME factor values scored on + # the 14:51 VWAP exec-to-exec return instead of close-to-close. The reports + # above are the close_to_close control and are left untouched. + exec_basis = run_exec_basis_evaluation( + factor_series, + spec, + eval_cfg, + book_processed, + cfg=cfg, + panel=panel, + symbols=symbols, + logger=logger, + report_dir=Path(cfg.output.report_dir), + stem=_REPORT_STEM, + ) + no_book_metrics = extract_metrics(reports.no_book) with_book_metrics = extract_metrics(reports.with_book) logger.info( @@ -491,6 +509,7 @@ def run_eval_peak_interval_kurtosis(config_path: str) -> PeakIntervalKurtosisEva no_book_metrics=no_book_metrics, with_book_metrics=with_book_metrics, reports=reports, + exec_basis=exec_basis, log_path=log_path, elapsed=time.monotonic() - started, ) diff --git a/qt/eval_peak_ridge_amount_ratio.py b/qt/eval_peak_ridge_amount_ratio.py index 472b76a..ca50f08 100644 --- a/qt/eval_peak_ridge_amount_ratio.py +++ b/qt/eval_peak_ridge_amount_ratio.py @@ -92,6 +92,7 @@ from factors.compute.intraday_derived import PeakRidgeAmountRatioFactor from factors.spec import FactorSpec from qt.config import RootConfig, load_config +from qt.exec_basis_eval import ExecBasisEvaluation, run_exec_basis_evaluation from qt.pipeline import ( _build_cache, _build_universe, @@ -580,6 +581,7 @@ class PeakRidgeAmountRatioEvalResult: no_book_metrics: dict with_book_metrics: dict reports: _RunReports + exec_basis: ExecBasisEvaluation log_path: Path elapsed: float @@ -677,6 +679,22 @@ def run_eval_peak_ridge_amount_ratio(config_path: str) -> PeakRidgeAmountRatioEv report_dir=Path(cfg.output.report_dir), ) + # Second evaluation basis (task card §2.3): the SAME factor values scored on + # the 14:51 VWAP exec-to-exec return instead of close-to-close. The reports + # above are the close_to_close control and are left untouched. + exec_basis = run_exec_basis_evaluation( + factor_series, + spec, + eval_cfg, + book_processed, + cfg=cfg, + panel=panel, + symbols=symbols, + logger=logger, + report_dir=Path(cfg.output.report_dir), + stem=_REPORT_STEM, + ) + no_book_metrics = extract_metrics(reports.no_book) with_book_metrics = extract_metrics(reports.with_book) logger.info( @@ -698,6 +716,7 @@ def run_eval_peak_ridge_amount_ratio(config_path: str) -> PeakRidgeAmountRatioEv no_book_metrics=no_book_metrics, with_book_metrics=with_book_metrics, reports=reports, + exec_basis=exec_basis, log_path=log_path, elapsed=time.monotonic() - started, ) diff --git a/qt/eval_ridge_minute_return.py b/qt/eval_ridge_minute_return.py index 9312607..adaa704 100644 --- a/qt/eval_ridge_minute_return.py +++ b/qt/eval_ridge_minute_return.py @@ -91,6 +91,7 @@ from factors.compute.intraday_derived import RidgeMinuteReturnFactor from factors.spec import FactorSpec from qt.config import RootConfig, load_config +from qt.exec_basis_eval import ExecBasisEvaluation, run_exec_basis_evaluation from qt.pipeline import ( _build_cache, _build_universe, @@ -582,6 +583,7 @@ class RidgeMinuteReturnEvalResult: no_book_metrics: dict with_book_metrics: dict reports: _RunReports + exec_basis: ExecBasisEvaluation log_path: Path elapsed: float @@ -677,6 +679,22 @@ def run_eval_ridge_minute_return(config_path: str) -> RidgeMinuteReturnEvalResul report_dir=Path(cfg.output.report_dir), ) + # Second evaluation basis (task card §2.3): the SAME factor values scored on + # the 14:51 VWAP exec-to-exec return instead of close-to-close. The reports + # above are the close_to_close control and are left untouched. + exec_basis = run_exec_basis_evaluation( + factor_series, + spec, + eval_cfg, + book_processed, + cfg=cfg, + panel=panel, + symbols=symbols, + logger=logger, + report_dir=Path(cfg.output.report_dir), + stem=_REPORT_STEM, + ) + no_book_metrics = extract_metrics(reports.no_book) with_book_metrics = extract_metrics(reports.with_book) logger.info( @@ -704,6 +722,7 @@ def run_eval_ridge_minute_return(config_path: str) -> RidgeMinuteReturnEvalResul no_book_metrics=no_book_metrics, with_book_metrics=with_book_metrics, reports=reports, + exec_basis=exec_basis, log_path=log_path, elapsed=time.monotonic() - started, ) diff --git a/qt/eval_valley_price_quantile.py b/qt/eval_valley_price_quantile.py index 99cd152..aef1f8b 100644 --- a/qt/eval_valley_price_quantile.py +++ b/qt/eval_valley_price_quantile.py @@ -99,6 +99,7 @@ from factors.compute.intraday_derived import ValleyPriceQuantileFactor from factors.spec import FactorSpec from qt.config import RootConfig, load_config +from qt.exec_basis_eval import ExecBasisEvaluation, run_exec_basis_evaluation from qt.pipeline import ( _build_cache, _build_universe, @@ -528,6 +529,7 @@ class ValleyPriceQuantileEvalResult: no_book_metrics: dict with_book_metrics: dict reports: _RunReports + exec_basis: ExecBasisEvaluation log_path: Path elapsed: float @@ -625,6 +627,22 @@ def run_eval_valley_price_quantile(config_path: str) -> ValleyPriceQuantileEvalR report_dir=Path(cfg.output.report_dir), ) + # Second evaluation basis (task card §2.3): the SAME factor values scored on + # the 14:51 VWAP exec-to-exec return instead of close-to-close. The reports + # above are the close_to_close control and are left untouched. + exec_basis = run_exec_basis_evaluation( + factor_series, + spec, + eval_cfg, + book_processed, + cfg=cfg, + panel=panel, + symbols=symbols, + logger=logger, + report_dir=Path(cfg.output.report_dir), + stem=_REPORT_STEM, + ) + no_book_metrics = extract_metrics(reports.no_book) with_book_metrics = extract_metrics(reports.with_book) logger.info( @@ -657,6 +675,7 @@ def run_eval_valley_price_quantile(config_path: str) -> ValleyPriceQuantileEvalR no_book_metrics=no_book_metrics, with_book_metrics=with_book_metrics, reports=reports, + exec_basis=exec_basis, log_path=log_path, elapsed=time.monotonic() - started, ) diff --git a/qt/eval_valley_relative_vwap.py b/qt/eval_valley_relative_vwap.py index 0300cc0..01ba042 100644 --- a/qt/eval_valley_relative_vwap.py +++ b/qt/eval_valley_relative_vwap.py @@ -74,6 +74,7 @@ from factors.compute.intraday_derived import ValleyRelativeVwapFactor from factors.spec import FactorSpec from qt.config import RootConfig, load_config +from qt.exec_basis_eval import ExecBasisEvaluation, run_exec_basis_evaluation from qt.pipeline import ( _build_cache, _build_universe, @@ -382,6 +383,7 @@ class ValleyRelativeVwapEvalResult: no_book_metrics: dict with_book_metrics: dict reports: _RunReports + exec_basis: ExecBasisEvaluation log_path: Path elapsed: float @@ -477,6 +479,22 @@ def run_eval_valley_relative_vwap(config_path: str) -> ValleyRelativeVwapEvalRes report_dir=Path(cfg.output.report_dir), ) + # Second evaluation basis (task card §2.3): the SAME factor values scored on + # the 14:51 VWAP exec-to-exec return instead of close-to-close. The reports + # above are the close_to_close control and are left untouched. + exec_basis = run_exec_basis_evaluation( + factor_series, + spec, + eval_cfg, + book_processed, + cfg=cfg, + panel=panel, + symbols=symbols, + logger=logger, + report_dir=Path(cfg.output.report_dir), + stem=_REPORT_STEM, + ) + no_book_metrics = extract_metrics(reports.no_book) with_book_metrics = extract_metrics(reports.with_book) logger.info( @@ -497,6 +515,7 @@ def run_eval_valley_relative_vwap(config_path: str) -> ValleyRelativeVwapEvalRes no_book_metrics=no_book_metrics, with_book_metrics=with_book_metrics, reports=reports, + exec_basis=exec_basis, log_path=log_path, elapsed=time.monotonic() - started, ) diff --git a/qt/eval_valley_ridge_vwap_ratio.py b/qt/eval_valley_ridge_vwap_ratio.py index c225595..25fe0c8 100644 --- a/qt/eval_valley_ridge_vwap_ratio.py +++ b/qt/eval_valley_ridge_vwap_ratio.py @@ -85,6 +85,7 @@ from factors.compute.intraday_derived import ValleyRidgeVwapRatioFactor from factors.spec import FactorSpec from qt.config import RootConfig, load_config +from qt.exec_basis_eval import ExecBasisEvaluation, run_exec_basis_evaluation from qt.pipeline import ( _build_cache, _build_universe, @@ -541,6 +542,7 @@ class ValleyRidgeVwapRatioEvalResult: no_book_metrics: dict with_book_metrics: dict reports: _RunReports + exec_basis: ExecBasisEvaluation log_path: Path elapsed: float @@ -637,6 +639,22 @@ def run_eval_valley_ridge_vwap_ratio(config_path: str) -> ValleyRidgeVwapRatioEv report_dir=Path(cfg.output.report_dir), ) + # Second evaluation basis (task card §2.3): the SAME factor values scored on + # the 14:51 VWAP exec-to-exec return instead of close-to-close. The reports + # above are the close_to_close control and are left untouched. + exec_basis = run_exec_basis_evaluation( + factor_series, + spec, + eval_cfg, + book_processed, + cfg=cfg, + panel=panel, + symbols=symbols, + logger=logger, + report_dir=Path(cfg.output.report_dir), + stem=_REPORT_STEM, + ) + no_book_metrics = extract_metrics(reports.no_book) with_book_metrics = extract_metrics(reports.with_book) logger.info( @@ -658,6 +676,7 @@ def run_eval_valley_ridge_vwap_ratio(config_path: str) -> ValleyRidgeVwapRatioEv no_book_metrics=no_book_metrics, with_book_metrics=with_book_metrics, reports=reports, + exec_basis=exec_basis, log_path=log_path, elapsed=time.monotonic() - started, ) diff --git a/qt/eval_volume_peak_count.py b/qt/eval_volume_peak_count.py index 95dc47d..43ab2fb 100644 --- a/qt/eval_volume_peak_count.py +++ b/qt/eval_volume_peak_count.py @@ -61,6 +61,7 @@ from factors.compute.intraday_derived import VolumePeakCountFactor from factors.spec import FactorSpec from qt.config import RootConfig, load_config +from qt.exec_basis_eval import ExecBasisEvaluation, run_exec_basis_evaluation from qt.pipeline import ( _build_cache, _build_universe, @@ -365,6 +366,7 @@ class VolumePeakCountEvalResult: no_book_metrics: dict with_book_metrics: dict reports: _RunReports + exec_basis: ExecBasisEvaluation log_path: Path elapsed: float @@ -459,6 +461,22 @@ def run_eval_volume_peak_count(config_path: str) -> VolumePeakCountEvalResult: report_dir=Path(cfg.output.report_dir), ) + # Second evaluation basis (task card §2.3): the SAME factor values scored on + # the 14:51 VWAP exec-to-exec return instead of close-to-close. The reports + # above are the close_to_close control and are left untouched. + exec_basis = run_exec_basis_evaluation( + factor_series, + spec, + eval_cfg, + book_processed, + cfg=cfg, + panel=panel, + symbols=symbols, + logger=logger, + report_dir=Path(cfg.output.report_dir), + stem=_REPORT_STEM, + ) + no_book_metrics = extract_metrics(reports.no_book) with_book_metrics = extract_metrics(reports.with_book) logger.info( @@ -479,6 +497,7 @@ def run_eval_volume_peak_count(config_path: str) -> VolumePeakCountEvalResult: no_book_metrics=no_book_metrics, with_book_metrics=with_book_metrics, reports=reports, + exec_basis=exec_basis, log_path=log_path, elapsed=time.monotonic() - started, ) diff --git a/qt/exec_basis_eval.py b/qt/exec_basis_eval.py new file mode 100644 index 0000000..e829aa9 --- /dev/null +++ b/qt/exec_basis_eval.py @@ -0,0 +1,398 @@ +"""Evaluate a minute-derived factor a SECOND time, on the ``exec_to_exec`` basis. + +The eleven minute-factor runners each already evaluate their factor twice +(no-book / with-book) against ``close_to_close`` daily returns. This module adds +the execution-anchored pair: the SAME factor values, the SAME evaluator, the SAME +two book settings — only the forward return changes, from +``close(t+h)/close(t)`` to the 14:51 VWAP anchor described in +:mod:`qt.exec_forward_returns`. + +Everything is deliberately additive: + + * the existing ``{stem}_no_book`` / ``{stem}_with_book`` artifacts are the + close-to-close control and are NOT touched — the new reports are written as + ``{stem}_exec_no_book`` / ``{stem}_exec_with_book``; + * the factor is computed ONCE by the runner and evaluated four times; + * the frozen evaluator is used as-is. The exec context supplies + ``forward_returns`` and deliberately omits ``price_panel``, so if the returns + were ever absent the evaluator's guard raises instead of quietly measuring a + close-to-close return under an ``exec_to_exec`` label. + +DISCLOSURE. The execution parameters, the coverage loss by cause, the measured +``stk_mins`` live-call count and the sanity-check headline travel INSIDE each exec +report via ``EvalContext.execution_capacity`` — the contract's documented +pass-through for "execution facts measured elsewhere". The two keys that axis +reads for a verdict (``tradable`` / ``capacity_sufficient``) are deliberately NOT +supplied: they are not measured here, so the Tradable axis stays NOT_ASSESSED, +exactly as it is on the close-to-close runs. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass +from pathlib import Path + +import pandas as pd + +from analytics.eval import ( + EvalConfig, + EvalContext, + FactorEvalReport, + StandardFactorEvaluator, +) +from analytics.eval.figures import render_factor_dashboard +from analytics.factor import forward_returns as _close_forward_returns +from data.clean.schema import DATE_LEVEL +from factors.spec import FactorSpec +from qt.config import RootConfig +from qt.exec_basis_sanity import ExecBasisSanity, check_exec_basis, render_sanity_report +from qt.exec_forward_returns import ( + ExecBasisParams, + build_exec_price_panel, + coverage_loss, + exec_forward_returns, + intraday_spec_variant, +) + + +@dataclass(frozen=True) +class ExecBasisEvaluation: + """The exec-basis half of a factor's evaluation (immutable).""" + + spec: FactorSpec + params: ExecBasisParams + artifact_path: Path + artifact_key: str + artifact_reused: bool + minute_live_calls: int + coverage: dict[str, object] + sanity: ExecBasisSanity + sanity_report_path: Path + no_book: FactorEvalReport + with_book: FactorEvalReport + no_book_md: Path + no_book_json: Path + with_book_md: Path + with_book_json: Path + no_book_dashboard: Path + with_book_dashboard: Path + no_book_metrics: dict + with_book_metrics: dict + elapsed: float + + +def _flatten_coverage(coverage: dict[str, object]) -> dict[str, object]: + """Flatten the coverage dict so every number renders as its own report row.""" + out: dict[str, object] = {} + for key, value in coverage.items(): + if isinstance(value, dict): + for cause, count in value.items(): + out[f"coverage_{key}_{cause}"] = count + else: + out[f"coverage_{key}"] = value + return out + + +def build_disclosure( + params: ExecBasisParams, + coverage: dict[str, object], + sanity: ExecBasisSanity, + *, + horizon: int, + artifact_path: Path, + artifact_key: str, + artifact_reused: bool, + minute_live_calls: int, + sanity_report_path: Path, +) -> dict[str, object]: + """The facts every exec report must carry (task card §2.4). + + Deliberately free of ``tradable`` / ``capacity_sufficient``: this run measures + a RETURN BASIS, not fill feasibility or capacity, and a verdict axis must not + move because a disclosure was added. + """ + disclosure: dict[str, object] = { + "return_basis": "exec_to_exec", + "forward_return_horizon_periods": horizon, + "decision_cutoff": params.decision_cutoff, + "data_lag": params.data_lag, + "session_open": params.session_open, + "execution_model": params.execution_model, + "execution_window": ( + f"[{params.execution_window[0]},{params.execution_window[1]}]" + ), + "execution_price_basis": params.execution_price_basis, + "execution_price_definition": "selected 1min bar amount / volume (RAW)", + "price_adjustment_applied": ( + "(raw_exec*adj_factor)_exit / (raw_exec*adj_factor)_entry - 1" + ), + "execution_parameter_source": params.source, + "stk_mins_live_calls": minute_live_calls, + "exec_price_artifact": str(artifact_path), + "exec_price_artifact_key": artifact_key, + "exec_price_artifact_reused": artifact_reused, + "sanity_report": str(sanity_report_path), + "missing_policy": ( + "no bar / undefined VWAP / unusable adj_factor -> NaN, counted by cause; " + "never a bar-close, daily-close or adj_factor=1.0 fallback" + ), + # The coverage loss is SMALL but it is not random, and the two must not be + # confused. Stated here so the percentage below can never stand alone and + # read as a rounding loss. + "coverage_bias_bad_vwap": ( + "LIQUIDITY-CORRELATED, NON-RANDOM exclusion: the dropped pairs are " + "exactly the execution minutes with no traded volume/amount. Small is " + "not the same as random." + ), + } + disclosure.update(_flatten_coverage(coverage)) + disclosure.update(sanity.headline()) + return disclosure + + +def _write_report(report: FactorEvalReport, report_dir: Path, stem: str) -> tuple[Path, Path]: + md_path = report_dir / f"{stem}.md" + json_path = report_dir / f"{stem}.json" + md_path.write_text(report.render(), encoding="utf-8") + json_path.write_text(report.to_json(), encoding="utf-8") + return md_path, json_path + + +def _extract_metrics(report: FactorEvalReport) -> dict: + """Headline verdict + gated metrics, mirroring the runners' ``extract_metrics``.""" + verdict = report.require_verdict() + sections = report.by_name() + + def payload(name: str) -> dict: + return dict(getattr(sections.get(name), "payload", {}) or {}) + + pred = payload("predictive_power") + coverage = payload("data_coverage") + incr = payload("purity") + return { + "deployment": verdict.verdict, + "predictive": verdict.predictive.verdict, + "incremental": verdict.incremental.verdict, + "tradable": verdict.tradable.verdict, + "ic_mean": pred.get("ic_mean"), + "ic_ir": pred.get("ic_ir"), + "ic_win_rate": pred.get("ic_win_rate"), + "ic_nw_t": pred.get("ic_nw_t"), + "settled_rebalances": coverage.get("settled_rebalances"), + "effective_samples": coverage.get("effective_samples"), + "span_days": coverage.get("span_days"), + "incremental_ic_ir": incr.get("incremental_ic_ir"), + "incremental_ic_mean": incr.get("incremental_ic_mean"), + } + + +def run_exec_basis_evaluation( + factor_panel: pd.Series | pd.DataFrame, + spec: FactorSpec, + eval_cfg: EvalConfig, + book: pd.DataFrame, + *, + cfg: RootConfig, + panel: pd.DataFrame, + symbols: list[str], + logger, + report_dir: Path, + stem: str, + force_rebuild: bool = False, +) -> ExecBasisEvaluation: + """Build the exec-to-exec returns, sanity-check them, evaluate twice, report. + + ``panel`` is the front-adjusted daily panel (it carries the raw ``adj_factor`` + the adjustment identity needs and defines the shared price grid); ``spec`` is + the factor's DAILY spec, from which the intraday variant is derived. + """ + started = time.monotonic() + params = ExecBasisParams.from_config(cfg) + exec_spec = intraday_spec_variant(spec, params) + horizon = spec.forward_return_horizon + + prices = build_exec_price_panel( + cfg, panel, symbols, params, logger, force_rebuild=force_rebuild + ) + + factor = ( + factor_panel + if isinstance(factor_panel, pd.Series) + else factor_panel[spec.factor_id] + ) + dates = pd.Index( + pd.unique(factor.index.get_level_values(DATE_LEVEL)), name=DATE_LEVEL + ).sort_values() + + exec_returns = exec_forward_returns(prices.adjusted_price, dates, horizon) + # The control series, derived by the SAME boundary the close_to_close run uses, + # restricted to the same grid first — so the comparison below is like-for-like. + on_grid = panel[panel.index.get_level_values(DATE_LEVEL).isin(dates)] + close_returns = _close_forward_returns(on_grid, periods=(horizon,))[ + f"forward_return_{horizon}d" + ] + coverage = coverage_loss( + exec_returns, + close_returns, + prices.frame["status"], + horizon, + ) + logger.info( + "exec basis coverage: %d/%d pairs measurable vs close_to_close " + "(lost %d = %.2f%%; no_bar=%d bad_vwap=%d bad_adj_factor=%d; %d symbols)", + coverage["exec_to_exec_measurable_pairs"], + coverage["close_to_close_measurable_pairs"], + coverage["lost_pairs"], + coverage["lost_pairs_pct_of_close_to_close"], + coverage["lost_pairs_by_cause"]["no_bar"], + coverage["lost_pairs_by_cause"]["bad_vwap"], + coverage["lost_pairs_by_cause"]["bad_adj_factor"], + coverage["distinct_symbols_affected"], + ) + + sanity = check_exec_basis( + prices.frame, + exec_returns, + close_returns, + params, + cfg.data.cache.root_dir, + horizon, + ) + logger.info( + "exec basis sanity: corr median=%.4f (p10=%.4f p90=%.4f over %d dates), " + "exec std=%.4f vs close std=%.4f, hand-check max diff=%.3e", + sanity.corr_median, sanity.corr_p10, sanity.corr_p90, sanity.corr_dates, + sanity.exec_stats["std"], sanity.close_stats["std"], + sanity.hand_check_max_abs_diff, + ) + + report_dir.mkdir(parents=True, exist_ok=True) + sanity_path = report_dir / f"{stem}_exec_basis_sanity.md" + sanity_path.write_text( + render_sanity_report( + sanity, + params, + coverage, + key=prices.key, + artifact_path=str(prices.path), + horizon=horizon, + minute_live_calls=prices.minute_live_calls, + ), + encoding="utf-8", + ) + + disclosure = build_disclosure( + params, + coverage, + sanity, + horizon=horizon, + artifact_path=prices.path, + artifact_key=prices.key, + artifact_reused=prices.reused, + minute_live_calls=prices.minute_live_calls, + sanity_report_path=sanity_path, + ) + + evaluator = StandardFactorEvaluator() + # NOTE: no price_panel. The exec basis is not derivable from a close panel, and + # withholding it means an absent forward_returns raises instead of silently + # falling back to close_to_close under an exec_to_exec label. + ctx_no_book = EvalContext( + forward_returns=exec_returns, + universe_symbols=tuple(symbols), + fee_rate=float(cfg.cost.fee_rate), + execution_capacity=disclosure, + ) + report_no_book, ir_no_book = evaluator.evaluate_with_ir( + factor_panel, exec_spec, eval_cfg, ctx_no_book + ) + ctx_with_book = EvalContext( + forward_returns=exec_returns, + universe_symbols=tuple(symbols), + fee_rate=float(cfg.cost.fee_rate), + known_factors=book, + execution_capacity=disclosure, + ) + report_with_book, ir_with_book = evaluator.evaluate_with_ir( + factor_panel, exec_spec, eval_cfg, ctx_with_book + ) + + nb_md, nb_json = _write_report(report_no_book, report_dir, f"{stem}_exec_no_book") + wb_md, wb_json = _write_report( + report_with_book, report_dir, f"{stem}_exec_with_book" + ) + nb_png = render_factor_dashboard( + report_no_book, ir_no_book, report_dir / f"{stem}_exec_no_book_dashboard.png" + ) + wb_png = render_factor_dashboard( + report_with_book, ir_with_book, report_dir / f"{stem}_exec_with_book_dashboard.png" + ) + return ExecBasisEvaluation( + spec=exec_spec, + params=params, + artifact_path=prices.path, + artifact_key=prices.key, + artifact_reused=prices.reused, + minute_live_calls=prices.minute_live_calls, + coverage=coverage, + sanity=sanity, + sanity_report_path=sanity_path, + no_book=report_no_book, + with_book=report_with_book, + no_book_md=nb_md, + no_book_json=nb_json, + with_book_md=wb_md, + with_book_json=wb_json, + no_book_dashboard=nb_png, + with_book_dashboard=wb_png, + no_book_metrics=_extract_metrics(report_no_book), + with_book_metrics=_extract_metrics(report_with_book), + elapsed=time.monotonic() - started, + ) + + +def _fmt(value: object, spec: str = ".4f") -> str: + """Format a metric a Skipped section may legitimately leave absent. + + The print happens AFTER the reports are written and outside the command's + error handling, so a None must render as "n/a" rather than raise a TypeError + that would bury four finished reports under a traceback. + """ + if value is None: + return "n/a" + try: + return format(value, spec) + except (TypeError, ValueError): + return str(value) + + +def format_exec_basis_line(result: ExecBasisEvaluation) -> str: + """The CLI's exec-basis summary — same shape as the close_to_close lines.""" + nb, wb = result.no_book_metrics, result.with_book_metrics + cov = result.coverage + causes = cov["lost_pairs_by_cause"] + return ( + f"exec-to-exec basis (14:51 {result.params.execution_price_basis}, adjusted; " + f"stk_mins_live_calls={result.minute_live_calls}):\n" + f" coverage loss vs close_to_close: {cov['lost_pairs']} pairs " + f"({_fmt(cov['lost_pairs_pct_of_close_to_close'], '.2f')}%) — " + f"no_bar={causes['no_bar']} bad_vwap={causes['bad_vwap']} " + f"bad_adj_factor={causes['bad_adj_factor']}, " + f"{cov['distinct_symbols_affected']} symbols; " + f"corr vs close median={_fmt(result.sanity.corr_median)}\n" + f" exec no-book: {nb['deployment']} (predictive={nb['predictive']}) " + f"ic_mean={_fmt(nb['ic_mean'])} ic_ir={_fmt(nb['ic_ir'], '.3f')}\n" + f" exec with-book: {wb['deployment']} (incremental={wb['incremental']}) " + f"incr_ic_ir={_fmt(wb['incremental_ic_ir'], '.3f')}\n" + f" exec reports: {result.no_book_md} | {result.with_book_md}\n" + f" sanity: {result.sanity_report_path}" + ) + + +__all__ = [ + "ExecBasisEvaluation", + "build_disclosure", + "format_exec_basis_line", + "run_exec_basis_evaluation", +] diff --git a/qt/exec_basis_sanity.py b/qt/exec_basis_sanity.py new file mode 100644 index 0000000..ca57094 --- /dev/null +++ b/qt/exec_basis_sanity.py @@ -0,0 +1,504 @@ +"""Implementation-INDEPENDENT sanity checks on the ``exec_to_exec`` return series. + +A wrong return series does not announce itself: every factor scored against it +still produces a plausible-looking IC, a plausible-looking quantile spread and a +confident verdict. Unit tests cannot catch this on their own — they check the code +against the same understanding that wrote it. So these four checks are deliberately +built from a DIFFERENT direction: + + 1. **Agreement.** Two one-day return series for the same stocks must move + together. The per-date cross-sectional correlation against ``close_to_close`` + should sit far above :data:`CLOSE_CORR_FLOOR`; a median below it means the + series is not measuring what it claims, and :func:`check_exec_basis` RAISES + rather than letting eleven factors be scored against it. + 2. **Magnitude.** The distribution (mean / std / p1 / p99) must be the same order + as the daily cross-section, not 10x or 0.1x it. + 3. **Hand computation.** Five (date, symbol) pairs are recomputed straight from + the raw cached 1min parquet with plain arithmetic — this module never calls + ``resolve_fill`` / ``bar_execution_price`` / ``build_execution_prices``, so it + cannot inherit a bug from them. Agreement to 1e-9 is checked and the five rows + are published so a reader can redo them. + 4. **Ex-date.** At least one holding period spanning a corporate action, shown + BOTH ways: the raw VWAP ratio (which reads the mechanical drop as a loss) and + the adjusted return (which does not). + +Everything here is read-only: no cache is warmed, no artifact is rewritten. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +import numpy as np +import pandas as pd + +from analytics.eval.ir import cross_section_corr +from data.cache.intraday_cache import ENDPOINT as MINUTE_ENDPOINT +from data.cache.intraday_parquet_store import IntradayParquetStore +from data.clean.intraday_schema import RAW_INTRADAY_FREQ +from data.clean.schema import DATE_LEVEL, SYMBOL_LEVEL +from qt.exec_forward_returns import ExecBasisParams + +#: Median per-date cross-sectional correlation with ``close_to_close`` below which +#: the exec series is treated as broken. Two one-day return series over the same +#: names differ only by their intraday measurement point; ~0.9 is a floor, not a +#: target (the observed value should be well above it). +CLOSE_CORR_FLOOR = 0.90 + +#: How many pairs get hand-recomputed. Five is what a reader will actually check. +HAND_CHECK_ROWS = 5 + +#: Tolerance for "the hand computation agrees" (double-precision round-trip slack). +HAND_CHECK_TOL = 1e-9 + +#: Deterministic sample: the same five rows every run, so the published check is +#: reproducible rather than a different lottery each time. +HAND_CHECK_SEED = 20260721 + + +@dataclass(frozen=True) +class ExecBasisSanity: + """The four §3 checks, as facts.""" + + corr_median: float + corr_p10: float + corr_p90: float + corr_dates: int + corr_median_spearman: float + exec_stats: dict[str, float] + close_stats: dict[str, float] + hand_checks: tuple[dict, ...] + hand_check_max_abs_diff: float + ex_date_checks: tuple[dict, ...] + corr_floor: float = CLOSE_CORR_FLOOR + + @property + def corr_ok(self) -> bool: + return bool(np.isfinite(self.corr_median) and self.corr_median >= self.corr_floor) + + def headline(self) -> dict[str, object]: + """The compact form that travels INSIDE every exec-basis report.""" + return { + "sanity_corr_vs_close_to_close_median": self.corr_median, + "sanity_corr_vs_close_to_close_p10": self.corr_p10, + "sanity_corr_vs_close_to_close_p90": self.corr_p90, + "sanity_corr_spearman_median": self.corr_median_spearman, + "sanity_corr_dates": self.corr_dates, + "sanity_corr_floor": self.corr_floor, + "sanity_exec_return_mean": self.exec_stats["mean"], + "sanity_exec_return_std": self.exec_stats["std"], + "sanity_exec_return_p1": self.exec_stats["p1"], + "sanity_exec_return_p99": self.exec_stats["p99"], + "sanity_close_return_mean": self.close_stats["mean"], + "sanity_close_return_std": self.close_stats["std"], + "sanity_close_return_p1": self.close_stats["p1"], + "sanity_close_return_p99": self.close_stats["p99"], + "sanity_hand_checked_rows": len(self.hand_checks), + "sanity_hand_check_max_abs_diff": self.hand_check_max_abs_diff, + "sanity_ex_date_pairs_shown": len(self.ex_date_checks), + } + + +def _stats(series: pd.Series) -> dict[str, float]: + values = series.to_numpy(dtype=float) + values = values[np.isfinite(values)] + if values.size == 0: + return {k: float("nan") for k in ("mean", "std", "p1", "p99", "n")} + return { + "mean": float(np.mean(values)), + "std": float(np.std(values, ddof=1)) if values.size > 1 else float("nan"), + "p1": float(np.percentile(values, 1)), + "p99": float(np.percentile(values, 99)), + "n": float(values.size), + } + + +def exit_anchor_dates(index: pd.MultiIndex, horizon: int) -> pd.Series: + """Per row, the date its forward return is measured AT — ``h`` periods later. + + Uses the same positional ``groupby(symbol).shift(-h)`` the return itself uses, + so an anchor reported here is the anchor the return actually used. + """ + dates = pd.Series(index.get_level_values(DATE_LEVEL), index=index) + return dates.groupby(level=SYMBOL_LEVEL, sort=False, group_keys=False).shift( + -horizon + ) + + +def _hand_price( + store: IntradayParquetStore, + symbol: str, + date: pd.Timestamp, + params: ExecBasisParams, +) -> dict[str, object]: + """Recompute one execution bar's VWAP from the raw parquet, by hand. + + Plain pandas + arithmetic on the STORED columns: pick the bars whose + ``bar_end`` falls in the window, take the earliest, divide ``amount`` by + ``volume``. No execution-layer helper is imported, on purpose. + """ + day = pd.Timestamp(date).normalize() + stored = store.read_range( + MINUTE_ENDPOINT, + symbol, + RAW_INTRADAY_FREQ, + day, + day + pd.Timedelta("23:59:59"), + ) + lo = day + pd.Timedelta(params.execution_window[0]) + hi = day + pd.Timedelta(params.execution_window[1]) + bar_end = pd.to_datetime(stored["bar_end"]) if len(stored) else pd.Series(dtype="datetime64[ns]") + window = stored.loc[(bar_end >= lo) & (bar_end <= hi)] if len(stored) else stored + if window.empty: + return {"bar_time": None, "amount": float("nan"), "volume": float("nan"), + "vwap": float("nan")} + row = window.sort_values("bar_end").iloc[0] + amount = float(row["amount"]) + volume = float(row["volume"]) + vwap = amount / volume if volume > 0 else float("nan") + return { + "bar_time": pd.Timestamp(row["bar_end"]), + "amount": amount, + "volume": volume, + "vwap": vwap, + } + + +def _hand_check_rows( + frame: pd.DataFrame, + exec_returns: pd.Series, + exits: pd.Series, + store: IntradayParquetStore, + params: ExecBasisParams, + n_rows: int, +) -> tuple[list[dict], float]: + """Recompute ``n_rows`` returns end-to-end from raw bars and compare.""" + finite = exec_returns[np.isfinite(exec_returns.to_numpy(dtype=float))] + if finite.empty: + return [], float("nan") + rng = np.random.default_rng(HAND_CHECK_SEED) + take = min(n_rows, len(finite)) + positions = rng.choice(len(finite), size=take, replace=False) + positions.sort() + + rows: list[dict] = [] + worst = 0.0 + for pos in positions: + key = finite.index[pos] + entry_date, symbol = pd.Timestamp(key[0]).normalize(), str(key[1]) + exit_date = pd.Timestamp(exits.loc[key]) + entry = _hand_price(store, symbol, entry_date, params) + exit_ = _hand_price(store, symbol, exit_date, params) + af_entry = float(frame.loc[(entry_date, symbol), "adj_factor"]) + af_exit = float(frame.loc[(exit_date, symbol), "adj_factor"]) + hand = (exit_["vwap"] * af_exit) / (entry["vwap"] * af_entry) - 1.0 + generated = float(finite.iloc[pos]) + diff = abs(hand - generated) + worst = max(worst, diff if np.isfinite(diff) else float("inf")) + rows.append( + { + "symbol": symbol, + "entry_date": str(entry_date.date()), + "entry_bar": str(entry["bar_time"]), + "entry_amount": entry["amount"], + "entry_volume": entry["volume"], + "entry_vwap": entry["vwap"], + "entry_adj_factor": af_entry, + "exit_date": str(exit_date.date()), + "exit_bar": str(exit_["bar_time"]), + "exit_amount": exit_["amount"], + "exit_volume": exit_["volume"], + "exit_vwap": exit_["vwap"], + "exit_adj_factor": af_exit, + "hand_return": hand, + "generated_return": generated, + "abs_diff": diff, + } + ) + return rows, worst + + +def _ex_date_rows( + frame: pd.DataFrame, + exec_returns: pd.Series, + exits: pd.Series, + n_rows: int, +) -> list[dict]: + """Holding periods that straddle a corporate action, shown adjusted and not.""" + finite = exec_returns[np.isfinite(exec_returns.to_numpy(dtype=float))] + if finite.empty: + return [] + af = frame["adj_factor"] + entry_af = af.reindex(finite.index).to_numpy(dtype=float) + exit_keys = pd.MultiIndex.from_arrays( + [ + pd.DatetimeIndex(exits.reindex(finite.index).to_numpy()), + finite.index.get_level_values(SYMBOL_LEVEL), + ], + names=[DATE_LEVEL, SYMBOL_LEVEL], + ) + exit_af = af.reindex(exit_keys).to_numpy(dtype=float) + raw = frame["raw_exec_price"] + entry_raw = raw.reindex(finite.index).to_numpy(dtype=float) + exit_raw = raw.reindex(exit_keys).to_numpy(dtype=float) + + ratio = np.where(np.isfinite(entry_af) & (entry_af > 0), exit_af / entry_af, np.nan) + moved = np.isfinite(ratio) & (np.abs(ratio - 1.0) > 1e-9) + if not moved.any(): + return [] + order = np.argsort(-np.abs(ratio - 1.0)) + rows: list[dict] = [] + for pos in order: + if not moved[pos]: + break + key = finite.index[pos] + unadjusted = exit_raw[pos] / entry_raw[pos] - 1.0 + rows.append( + { + "symbol": str(key[1]), + "entry_date": str(pd.Timestamp(key[0]).date()), + "exit_date": str(pd.Timestamp(exit_keys[pos][0]).date()), + "entry_raw_vwap": float(entry_raw[pos]), + "exit_raw_vwap": float(exit_raw[pos]), + "entry_adj_factor": float(entry_af[pos]), + "exit_adj_factor": float(exit_af[pos]), + "unadjusted_return": float(unadjusted), + "adjusted_return": float(finite.iloc[pos]), + "correction": float(finite.iloc[pos] - unadjusted), + } + ) + if len(rows) >= n_rows: + break + return rows + + +def check_exec_basis( + frame: pd.DataFrame, + exec_returns: pd.Series, + close_returns: pd.Series, + params: ExecBasisParams, + cache_root: str, + horizon: int, + *, + n_hand_checks: int = HAND_CHECK_ROWS, + n_ex_date_checks: int = 1, + corr_floor: float = CLOSE_CORR_FLOOR, + enforce: bool = True, +) -> ExecBasisSanity: + """Run the four checks; RAISE when the agreement check fails. + + ``enforce=False`` is for tests that need to inspect a deliberately broken + series. A production caller leaves it True: a return basis that disagrees with + close-to-close is a bug signal, and the eleven verdicts that would be built on + it are worth less than nothing. + """ + aligned_close = close_returns.reindex(exec_returns.index) + ic_dates = pd.Index( + pd.unique(exec_returns.index.get_level_values(DATE_LEVEL)), name=DATE_LEVEL + ).sort_values() + pearson = cross_section_corr( + exec_returns, aligned_close, rank=False, dates=ic_dates + ) + spearman = cross_section_corr( + exec_returns, aligned_close, rank=True, dates=ic_dates + ) + usable = pearson.dropna() + + exits = exit_anchor_dates(exec_returns.index, horizon) + store = IntradayParquetStore(cache_root) + hand_rows, worst = _hand_check_rows( + frame, exec_returns, exits, store, params, n_hand_checks + ) + ex_rows = _ex_date_rows(frame, exec_returns, exits, n_ex_date_checks) + + sanity = ExecBasisSanity( + corr_median=float(usable.median()) if len(usable) else float("nan"), + corr_p10=float(usable.quantile(0.10)) if len(usable) else float("nan"), + corr_p90=float(usable.quantile(0.90)) if len(usable) else float("nan"), + corr_dates=int(len(usable)), + corr_median_spearman=( + float(spearman.dropna().median()) if spearman.notna().any() else float("nan") + ), + exec_stats=_stats(exec_returns), + close_stats=_stats(aligned_close), + hand_checks=tuple(hand_rows), + hand_check_max_abs_diff=worst, + ex_date_checks=tuple(ex_rows), + corr_floor=corr_floor, + ) + if enforce and not sanity.corr_ok: + raise ValueError( + f"exec_to_exec sanity check FAILED: the median per-date cross-sectional " + f"correlation with close_to_close is {sanity.corr_median:.4f} over " + f"{sanity.corr_dates} dates, below the {corr_floor:.2f} floor. Two " + f"one-day return series over the same names cannot disagree this much " + f"unless one of them is wrong — most likely the execution bar, the VWAP, " + f"the adjustment identity or the h-step alignment. STOP and investigate; " + f"do NOT explain this away and do not score factors against it." + ) + if enforce and np.isfinite(worst) and worst > HAND_CHECK_TOL and hand_rows: + raise ValueError( + f"exec_to_exec sanity check FAILED: a hand recomputation from the raw " + f"cached 1min bars disagrees with the generated return by {worst:.3e} " + f"(tolerance {HAND_CHECK_TOL:.0e}). The generated series does not equal " + f"(raw VWAP * adj_factor) exit / entry - 1 for at least one pair." + ) + return sanity + + +def _fmt(value: object, digits: int = 6) -> str: + if isinstance(value, float): + return "n/a" if not np.isfinite(value) else f"{value:.{digits}f}" + return "n/a" if value is None else str(value) + + +def render_sanity_report( + sanity: ExecBasisSanity, + params: ExecBasisParams, + coverage: dict[str, object], + *, + key: str, + artifact_path: str, + horizon: int, + minute_live_calls: int, +) -> str: + """Deterministic Markdown for the four checks — the audit trail for §3.""" + lines: list[str] = [] + lines.append("# exec-to-exec return basis — sanity checks") + lines.append("") + lines.append( + "The eleven minute-derived factors are scored against the execution-anchored " + "return below. These checks exist because a wrong return series still " + "produces confident-looking verdicts." + ) + lines.append("") + lines.append("## Basis and execution parameters") + lines.append("") + lines.append("| item | value |") + lines.append("|---|---|") + lines.append("| return basis | `exec_to_exec` |") + lines.append(f"| forward return horizon | {horizon} evaluation period(s) |") + lines.append(f"| signal cutoff (decision) | {params.decision_cutoff} |") + lines.append(f"| data lag | {params.data_lag} |") + lines.append(f"| session open | {params.session_open} |") + lines.append(f"| execution model | {params.execution_model} |") + lines.append( + f"| execution window | [{params.execution_window[0]}, " + f"{params.execution_window[1]}] |" + ) + lines.append(f"| price basis | {params.execution_price_basis} (bar amount/volume) |") + lines.append("| adjustment | `(raw*af)_exit / (raw*af)_entry - 1` (applied) |") + lines.append(f"| parameter source | {params.source} |") + lines.append(f"| artifact | `{artifact_path}` (key `{key}`) |") + lines.append(f"| stk_mins live calls | {minute_live_calls} |") + lines.append("") + + lines.append("## 1. Agreement with close_to_close") + lines.append("") + lines.append( + f"Per-date cross-sectional correlation over {sanity.corr_dates} dates — " + f"median **{_fmt(sanity.corr_median, 4)}** " + f"(p10 {_fmt(sanity.corr_p10, 4)}, p90 {_fmt(sanity.corr_p90, 4)}); " + f"Spearman median {_fmt(sanity.corr_median_spearman, 4)}. " + f"Floor {sanity.corr_floor:.2f} — " + f"{'PASS' if sanity.corr_ok else 'FAIL'}." + ) + lines.append("") + + lines.append("## 2. Magnitude") + lines.append("") + lines.append("| series | mean | std | p1 | p99 | n |") + lines.append("|---|---|---|---|---|---|") + for name, stats in (("exec_to_exec", sanity.exec_stats), ("close_to_close", sanity.close_stats)): + lines.append( + f"| {name} | {_fmt(stats['mean'])} | {_fmt(stats['std'])} | " + f"{_fmt(stats['p1'])} | {_fmt(stats['p99'])} | {_fmt(stats['n'], 0)} |" + ) + lines.append("") + + lines.append("## 3. Hand recomputation from the raw cached bars") + lines.append("") + lines.append( + f"{len(sanity.hand_checks)} pair(s), recomputed with plain arithmetic on the " + f"stored 1min rows (no execution-layer helper involved). Worst absolute " + f"difference **{_fmt(sanity.hand_check_max_abs_diff, 12)}**." + ) + lines.append("") + if sanity.hand_checks: + lines.append( + "| symbol | entry date | entry bar | amount | volume | vwap | af | " + "exit date | exit bar | vwap | af | hand return | generated | diff |" + ) + lines.append("|" + "---|" * 14) + for row in sanity.hand_checks: + lines.append( + f"| {row['symbol']} | {row['entry_date']} | {row['entry_bar']} | " + f"{_fmt(row['entry_amount'], 2)} | {_fmt(row['entry_volume'], 2)} | " + f"{_fmt(row['entry_vwap'], 6)} | {_fmt(row['entry_adj_factor'], 6)} | " + f"{row['exit_date']} | {row['exit_bar']} | " + f"{_fmt(row['exit_vwap'], 6)} | {_fmt(row['exit_adj_factor'], 6)} | " + f"{_fmt(row['hand_return'], 10)} | {_fmt(row['generated_return'], 10)} | " + f"{_fmt(row['abs_diff'], 12)} |" + ) + lines.append("") + + lines.append("## 4. Ex-date holding period") + lines.append("") + if not sanity.ex_date_checks: + lines.append( + "No holding period in this window straddles an `adj_factor` change, so " + "the adjustment cannot be demonstrated on real data here." + ) + else: + lines.append( + "`adj_factor` moves inside the holding period, so the raw VWAP ratio " + "carries a mechanical drop the adjusted return removes." + ) + lines.append("") + lines.append( + "| symbol | entry | exit | raw vwap in | raw vwap out | af in | af out | " + "unadjusted | adjusted | correction |" + ) + lines.append("|" + "---|" * 10) + for row in sanity.ex_date_checks: + lines.append( + f"| {row['symbol']} | {row['entry_date']} | {row['exit_date']} | " + f"{_fmt(row['entry_raw_vwap'])} | {_fmt(row['exit_raw_vwap'])} | " + f"{_fmt(row['entry_adj_factor'])} | {_fmt(row['exit_adj_factor'])} | " + f"{_fmt(row['unadjusted_return'])} | {_fmt(row['adjusted_return'])} | " + f"{_fmt(row['correction'])} |" + ) + lines.append("") + + lines.append("## Coverage loss vs close_to_close") + lines.append("") + lines.append( + "⚠️ `bad_vwap` is a **liquidity-correlated, non-random** exclusion, not a " + "rounding loss: the dropped pairs are exactly the execution minutes with no " + "traded volume or amount, i.e. the names that did not trade at the fill " + "time. A small percentage is not the same as a random one — read the number " + "below as a biased sample restriction with a direction." + ) + lines.append("") + lines.append("| item | value |") + lines.append("|---|---|") + for label, value in coverage.items(): + if isinstance(value, dict): + for cause, count in value.items(): + lines.append(f"| {label}.{cause} | {count} |") + else: + lines.append(f"| {label} | {_fmt(value, 4)} |") + lines.append("") + return "\n".join(lines) + + +__all__ = [ + "CLOSE_CORR_FLOOR", + "HAND_CHECK_ROWS", + "HAND_CHECK_SEED", + "HAND_CHECK_TOL", + "ExecBasisSanity", + "check_exec_basis", + "exit_anchor_dates", + "render_sanity_report", +] diff --git a/qt/exec_forward_returns.py b/qt/exec_forward_returns.py new file mode 100644 index 0000000..51e4e84 --- /dev/null +++ b/qt/exec_forward_returns.py @@ -0,0 +1,621 @@ +"""``exec_to_exec`` forward returns — the 14:51 execution-anchored evaluation basis. + +WHY THIS EXISTS. A minute-derived factor is fixed at the 14:50 signal cutoff, and +the only fill this framework can actually MODEL is the 1min bar in the execution +window that opens at 14:51. Scoring such a factor on ``close(t+h)/close(t)`` +credits it with a closing auction this project cannot simulate (the +``closing_call_proxy`` execution model needs ``stk_auction_*``, which the data +plan does not carry). This module builds the honest alternative:: + + R_exec[d, s] = (vwap(d+h, s) * af(d+h, s)) / (vwap(d, s) * af(d, s)) - 1 + + * ``vwap(d, s)`` is the RAW ``amount / volume`` of the EARLIEST 1min bar whose + ``bar_end`` falls in the execution window — the bar + :func:`runtime.intraday_execution.resolve_fill` selects, priced by + :func:`runtime.intraday_execution.bar_execution_price`. Both are REUSED, never + reimplemented: two copies of an execution rule drift, and this project has + paid for that lesson. + * ``af`` is the daily panel's raw cumulative ``adj_factor``. The per-symbol + anchor ``data/clean/adjust.py`` divides by cancels in the ratio, so a holding + period spanning an ex-date is corporate-action free WITHOUT re-deriving a qfq + series — the identity PR #75 established for + :meth:`runtime.backtest.event_models.IntradayTailEventModel.holding_returns`, + copied here rather than re-derived. + * ``d+h`` is ``h`` steps on the FACTOR'S OWN EVALUATION GRID (h is "a horizon in + evaluation periods", per ``FactorSpec.forward_return_horizon``), never the + next calendar day and never the next day the minute cache happens to hold. + +UNDEFINED IS MISSING — NEVER A FALLBACK. No bar in the window, a bar whose VWAP +is undefined (missing / non-finite / non-positive volume or amount), or a missing +/ non-positive ``adj_factor`` all yield NaN for that (date, symbol), counted under +their OWN cause (:data:`MISS_REASONS`). The bar close is never substituted for the +VWAP, the daily close is never substituted for the bar, and ``adj_factor`` is +never assumed to be 1.0 — each of those would silently manufacture a return. + +CACHE-ONLY. Minute bars are read straight from +:class:`data.cache.intraday_parquet_store.IntradayParquetStore`, which has no +fetch closure: a miss yields no rows. ``stk_mins`` live calls are therefore +provably zero, and that zero is reported rather than asserted. + +SHARED ARTIFACT. The expensive part — one adjusted execution price per +(date, symbol) — is written once to ``artifacts/data`` and reused by every factor +evaluated on the same universe / window / execution parameters, so the whole +factor family is scored against ONE return series and cross-factor comparisons +stay meaningful. The cache key encodes the universe, the window, the evaluation +grid and every execution parameter, so a changed parameter can never hit a stale +artifact. +""" + +from __future__ import annotations + +import hashlib +import json +import os +from dataclasses import dataclass, replace +from pathlib import Path + +import numpy as np +import pandas as pd + +from data.cache.intraday_cache import ENDPOINT as MINUTE_ENDPOINT +from data.cache.intraday_cache import READ_COLUMNS +from data.cache.intraday_parquet_store import IntradayParquetStore +from data.clean.intraday_schema import RAW_INTRADAY_FREQ, normalize_intraday_bars +from data.clean.schema import DATE_LEVEL, SYMBOL_LEVEL +from factors.spec import INTRADAY_RETURN_BASIS, FactorSpec +from qt.config import IntradayCfg, RootConfig +from runtime.intraday_execution import ( + REASON_NO_BAR, + IntradayExecutionConfig, + build_execution_prices, +) + +#: Bumped whenever the artifact's columns or their meaning change, so an old file +#: can never be read as a new one (it simply misses the key and is rebuilt). +ARTIFACT_SCHEMA_VERSION = "exec_prices_v1" + +#: Per-(date, symbol) outcome of pricing the execution bar. +STATUS_OK = "ok" +#: No 1min bar in the execution window (suspended, delisted, uncached day, ...). +MISS_NO_BAR = "no_bar" +#: A bar exists but its configured price basis is undefined (for ``bar_vwap``: +#: missing / non-finite / non-positive ``volume`` or ``amount``). +MISS_BAD_VWAP = "bad_vwap" +#: The bar priced fine, but the daily panel's ``adj_factor`` is absent, NaN or +#: non-positive, so no corporate-action-free price can be formed. +MISS_BAD_ADJ_FACTOR = "bad_adj_factor" +MISS_REASONS: tuple[str, ...] = (MISS_NO_BAR, MISS_BAD_VWAP, MISS_BAD_ADJ_FACTOR) + +#: Columns of the persisted artifact (stable; see ARTIFACT_SCHEMA_VERSION). +ARTIFACT_COLUMNS: tuple[str, ...] = ( + "raw_exec_price", + "adj_factor", + "adj_exec_price", + "exec_time", + "status", +) + + +def _sha1(text: str) -> str: + return hashlib.sha1(text.encode("utf-8")).hexdigest() # noqa: S324 - not crypto + + +# --------------------------------------------------------------------------- # +# The ONE source of the execution parameters +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class ExecBasisParams: + """Every execution parameter the ``exec_to_exec`` basis depends on. + + ONE object feeds all three consumers, so they cannot disagree: + + * :meth:`exec_config` — what actually selects and prices the bar; + * :meth:`spec_fields` — the five ``FactorSpec`` minute-block fields the + report publishes (a spec that DESCRIBED different parameters than the + returns were computed with would be a lie the contract cannot catch); + * :meth:`as_dict` — the artifact cache key, so changing a parameter forces a + rebuild instead of hitting a stale file. + + ``from_config`` reads the config's ``intraday`` block when it declares one and + otherwise falls back to :class:`qt.config.IntradayCfg`'s own defaults — the + project's declared minute conventions. Which of the two applied is recorded in + ``source`` and disclosed, never left to be assumed. + """ + + decision_cutoff: str + data_lag: str + session_open: str + execution_model: str + execution_window: tuple[str, str] + execution_price_basis: str + source: str + + @classmethod + def from_config(cls, cfg: RootConfig) -> ExecBasisParams: + declared = cfg.intraday is not None + ic = cfg.intraday if declared else IntradayCfg() + return cls( + decision_cutoff=ic.decision_time, + data_lag=ic.data_lag, + session_open=ic.session_open, + execution_model=ic.execution_model, + execution_window=(ic.execution_window[0], ic.execution_window[1]), + execution_price_basis=ic.execution_price_basis, + source=( + "config intraday block" + if declared + else "qt.config.IntradayCfg defaults (config declares no intraday block)" + ), + ) + + def exec_config(self) -> IntradayExecutionConfig: + """The runtime execution config — validated by ITS OWN ``__post_init__``. + + Building it here means the window/model/basis are checked by the same code + the intraday backtest uses; there is no second validation path to drift. + """ + return IntradayExecutionConfig( + decision_time=self.decision_cutoff, + data_lag=self.data_lag, + execution_model=self.execution_model, + execution_window=self.execution_window, + execution_price_basis=self.execution_price_basis, + ) + + def spec_fields(self) -> dict[str, str]: + """The five ``FactorSpec`` minute-block fields, from THESE parameters.""" + return { + "decision_cutoff": self.decision_cutoff, + "data_lag": self.data_lag, + "session_open": self.session_open, + "execution_model": self.execution_model, + "execution_window": ( + f"[{self.execution_window[0]},{self.execution_window[1]}]" + ), + } + + def as_dict(self) -> dict[str, object]: + """Deterministic, JSON-safe view for the cache key and the disclosure.""" + return { + "decision_cutoff": self.decision_cutoff, + "data_lag": self.data_lag, + "session_open": self.session_open, + "execution_model": self.execution_model, + "execution_window": list(self.execution_window), + "execution_price_basis": self.execution_price_basis, + } + + +def intraday_spec_variant(spec: FactorSpec, params: ExecBasisParams) -> FactorSpec: + """Derive the ``exec_to_exec`` twin of a daily ``spec`` — no literal editing. + + The factor itself is untouched (same ``factor_id`` / ``version`` / hypothesis / + horizon / inputs: the VALUES are identical, only what they are scored against + changes). ``FactorSpec.__post_init__`` then enforces the minute block's + completeness; that check is deliberately left to run rather than worked around. + """ + if spec.is_intraday: + raise ValueError( + f"intraday_spec_variant expects the DAILY spec of {spec.factor_id!r} " + f"(is_intraday=False); it is already an intraday variant. Deriving a " + f"variant of a variant would hide which basis the report describes." + ) + return replace( + spec, + is_intraday=True, + return_basis=INTRADAY_RETURN_BASIS, + **params.spec_fields(), + ) + + +# --------------------------------------------------------------------------- # +# The adjusted execution-price panel (the shared, cached artifact) +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class ExecPricePanel: + """One adjusted execution price per (date, symbol), plus how it was obtained.""" + + frame: pd.DataFrame # MultiIndex(date, symbol), ARTIFACT_COLUMNS + params: ExecBasisParams + key: str + path: Path + reused: bool + minute_live_calls: int + symbols_requested: int + symbols_with_bars: int + raw_minute_rows: int + + @property + def adjusted_price(self) -> pd.Series: + """``raw VWAP * adj_factor`` — NaN wherever the pair is not ``ok``.""" + return self.frame["adj_exec_price"] + + def status_counts(self) -> dict[str, int]: + """Row count per status over the whole (date, symbol) grid.""" + counts = self.frame["status"].value_counts() + keys = (STATUS_OK, *MISS_REASONS) + return {k: int(counts.get(k, 0)) for k in keys} + + +def _restrict_to_execution_window( + stored: pd.DataFrame, params: ExecBasisParams +) -> pd.DataFrame: + """Keep only bars whose ``bar_end`` time-of-day lies in the execution window. + + A PERFORMANCE pre-filter, not a semantic one: :func:`resolve_fill` applies the + very same window to whatever it is handed, so dropping the other ~230 bars per + day cannot change which bar is selected or how it is priced (locked by + ``test_window_prefilter_matches_full_day_build``). It is what makes a five-year + all-symbol build affordable — the alternative is normalizing ~40x more rows. + """ + if stored.empty: + return stored + bar_end = pd.to_datetime(stored["bar_end"]) + tod = bar_end - bar_end.dt.normalize() + lo = pd.Timedelta(params.execution_window[0]) + hi = pd.Timedelta(params.execution_window[1]) + return stored.loc[(tod >= lo) & (tod <= hi)] + + +def _symbol_exec_fills( + store: IntradayParquetStore, + symbol: str, + dates: list[pd.Timestamp], + start: pd.Timestamp, + end: pd.Timestamp, + params: ExecBasisParams, + exec_cfg: IntradayExecutionConfig, +) -> tuple[list, int]: + """``(fills, raw_minute_rows)`` for ONE symbol over ``dates`` — cache-only. + + ``IntradayParquetStore.read_range`` carries no fetch closure, so a cache miss + yields an empty frame and NEVER a live ``stk_mins`` call. + """ + stored = store.read_range(MINUTE_ENDPOINT, symbol, RAW_INTRADAY_FREQ, start, end) + raw_rows = int(len(stored)) + window = _restrict_to_execution_window(stored, params) + if window.empty: + return [], raw_rows + bars = normalize_intraday_bars( + window.rename(columns={"bar_end": "time"})[READ_COLUMNS], + freq=RAW_INTRADAY_FREQ, + data_lag=params.data_lag, + ) + _, fills = build_execution_prices(bars, dates, [symbol], exec_cfg) + return fills, raw_rows + + +def _fills_frame(fills: list, symbol: str) -> pd.DataFrame: + """Fills of one symbol -> frame indexed by (date, symbol).""" + if not fills: + return pd.DataFrame( + columns=["raw_exec_price", "exec_time", "miss_reason"], + index=pd.MultiIndex.from_arrays( + [pd.DatetimeIndex([]), pd.Index([], dtype=object)], + names=[DATE_LEVEL, SYMBOL_LEVEL], + ), + ) + index = pd.MultiIndex.from_arrays( + [ + pd.DatetimeIndex([pd.Timestamp(f.date).normalize() for f in fills]), + pd.Index([symbol] * len(fills), dtype=object), + ], + names=[DATE_LEVEL, SYMBOL_LEVEL], + ) + return pd.DataFrame( + { + "raw_exec_price": [ + np.nan if f.exec_price is None else float(f.exec_price) for f in fills + ], + "exec_time": [ + pd.NaT if f.exec_time is None else pd.Timestamp(f.exec_time) + for f in fills + ], + "miss_reason": [ + None + if not f.blocked + else (MISS_NO_BAR if f.reason == REASON_NO_BAR else MISS_BAD_VWAP) + for f in fills + ], + }, + index=index, + ) + + +def _assemble(panel: pd.DataFrame, collected: pd.DataFrame) -> pd.DataFrame: + """Join the per-symbol fills onto the panel grid and classify every row.""" + frame = pd.DataFrame(index=panel.index) + aligned = collected.reindex(panel.index) + frame["raw_exec_price"] = aligned["raw_exec_price"].astype(float) + frame["adj_factor"] = pd.to_numeric(panel["adj_factor"], errors="coerce") + frame["exec_time"] = pd.to_datetime(aligned["exec_time"]) + + price = frame["raw_exec_price"].to_numpy(dtype=float) + factor = frame["adj_factor"].to_numpy(dtype=float) + price_ok = np.isfinite(price) + # Mirrors IntradayTailEventModel._index_adj_factors: only a finite, STRICTLY + # POSITIVE factor is usable. Treating a missing one as 1.0 is precisely the + # defect PR #75 removed from the intraday holding return. + factor_ok = np.isfinite(factor) & (factor > 0.0) + + frame["adj_exec_price"] = np.where(price_ok & factor_ok, price * factor, np.nan) + + reason = aligned["miss_reason"].to_numpy(dtype=object) + # A symbol/date the loop never produced a fill for (no cached minute at all) + # is "no bar" — the same cause, reached without a fill record. + reason = np.where(pd.isna(reason), MISS_NO_BAR, reason) + status = np.where( + price_ok, + np.where(factor_ok, STATUS_OK, MISS_BAD_ADJ_FACTOR), + reason, + ) + frame["status"] = pd.Series(status, index=frame.index, dtype=object) + return frame[list(ARTIFACT_COLUMNS)] + + +def artifact_key( + cfg: RootConfig, + symbols: list[str], + dates: pd.Index, + params: ExecBasisParams, +) -> tuple[str, dict]: + """``(key, payload)`` identifying one adjusted-execution-price artifact. + + Everything the numbers depend on is in the payload: the universe, the window, + the cache root, the exact symbol set and evaluation grid, and every execution + parameter. Change any of them and the key changes, so a stale artifact cannot + be mistaken for a current one — the failure mode that would silently score a + factor family against the wrong returns. + """ + payload = { + "schema": ARTIFACT_SCHEMA_VERSION, + "universe_type": cfg.universe.type, + "index_code": cfg.universe.index_code, + "start": str(cfg.data.start), + "end": str(cfg.data.end), + "cache_root": str(cfg.data.cache.root_dir), + "price_adjust": "qfq", + "params": params.as_dict(), + "n_symbols": len(symbols), + "n_dates": int(len(dates)), + "symbols_sha1": _sha1("|".join(sorted(str(s) for s in symbols))), + "dates_sha1": _sha1( + "|".join(pd.Timestamp(d).strftime("%Y-%m-%d") for d in dates) + ), + } + blob = json.dumps(payload, sort_keys=True, separators=(",", ":")) + return _sha1(blob)[:16], payload + + +def _artifact_paths(cfg: RootConfig, key: str) -> tuple[Path, Path]: + stem = f"exec_forward_returns_{key}" + directory = Path(cfg.output.data_dir) + return directory / f"{stem}.parquet", directory / f"{stem}.json" + + +def _write_atomic(frame: pd.DataFrame, path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + f".tmp{os.getpid()}") + frame.to_parquet(tmp) + os.replace(tmp, path) + + +def build_exec_price_panel( + cfg: RootConfig, + panel: pd.DataFrame, + symbols: list[str], + params: ExecBasisParams, + logger, + *, + force_rebuild: bool = False, +) -> ExecPricePanel: + """Adjusted execution prices for the whole (date, symbol) grid — build or reuse. + + ``panel`` is the front-adjusted daily panel: it supplies BOTH the evaluation + grid (its index) and the raw ``adj_factor`` the adjustment identity needs. + Prices come only from the minute cache; nothing is fetched. + """ + if "adj_factor" not in panel.columns: + raise ValueError( + "build_exec_price_panel needs the daily panel's raw 'adj_factor' column " + "(front_adjust preserves it). Without it the exec-to-exec return cannot " + "be made corporate-action free, and assuming 1.0 is not an option." + ) + dates = pd.Index( + pd.unique(panel.index.get_level_values(DATE_LEVEL)), name=DATE_LEVEL + ).sort_values() + key, key_payload = artifact_key(cfg, symbols, dates, params) + parquet_path, meta_path = _artifact_paths(cfg, key) + + if parquet_path.exists() and not force_rebuild: + frame = pd.read_parquet(parquet_path) + logger.info( + "exec price artifact: REUSED %s (%d rows, key=%s)", + parquet_path, len(frame), key, + ) + priced = frame["raw_exec_price"].notna().groupby(level=SYMBOL_LEVEL).any() + return ExecPricePanel( + frame=frame, + params=params, + key=key, + path=parquet_path, + reused=True, + minute_live_calls=0, + symbols_requested=len(symbols), + symbols_with_bars=int(priced.sum()), + raw_minute_rows=0, + ) + + exec_cfg = params.exec_config() + store = IntradayParquetStore(cfg.data.cache.root_dir) + start = pd.Timestamp(dates[0]).normalize() + end = pd.Timestamp(dates[-1]).normalize() + pd.Timedelta("23:59:59") + + grid = pd.DataFrame( + { + "date": panel.index.get_level_values(DATE_LEVEL), + "symbol": panel.index.get_level_values(SYMBOL_LEVEL).astype(str), + } + ) + by_symbol: dict[str, list[pd.Timestamp]] = { + str(symbol): list(part["date"]) + for symbol, part in grid.groupby("symbol", sort=True) + } + + parts: list[pd.DataFrame] = [] + raw_rows = 0 + with_bars = 0 + ordered = sorted(by_symbol) + for i, symbol in enumerate(ordered): + fills, rows = _symbol_exec_fills( + store, symbol, by_symbol[symbol], start, end, params, exec_cfg + ) + raw_rows += rows + if fills: + part = _fills_frame(fills, symbol) + if part["raw_exec_price"].notna().any(): + with_bars += 1 + parts.append(part) + if (i + 1) % 100 == 0: + logger.info( + "exec price build: %d/%d symbols (%d with a priced bar)", + i + 1, len(ordered), with_bars, + ) + collected = ( + pd.concat(parts) + if parts + else _fills_frame([], "") + ) + if collected.index.has_duplicates: + raise ValueError( + "exec price build produced duplicate (date, symbol) rows; one would " + "silently win the alignment. This means a symbol was processed twice." + ) + frame = _assemble(panel, collected) + + _write_atomic(frame, parquet_path) + meta_path.parent.mkdir(parents=True, exist_ok=True) + meta_path.write_text( + json.dumps({"key": key, **key_payload}, indent=2, sort_keys=True), + encoding="utf-8", + ) + logger.info( + "exec price artifact: BUILT %s (%d rows, %d symbols with a priced bar, " + "%d raw 1min rows read, stk_mins_live_calls=0, key=%s)", + parquet_path, len(frame), with_bars, raw_rows, key, + ) + return ExecPricePanel( + frame=frame, + params=params, + key=key, + path=parquet_path, + reused=False, + minute_live_calls=0, + symbols_requested=len(symbols), + symbols_with_bars=with_bars, + raw_minute_rows=raw_rows, + ) + + +# --------------------------------------------------------------------------- # +# Forward returns on the FACTOR's own evaluation grid +# --------------------------------------------------------------------------- # +def exec_forward_returns( + adjusted_price: pd.Series, dates: pd.Index, horizon: int +) -> pd.Series: + """``R_h`` from adjusted execution prices, shifted on ``dates`` — h PERIODS. + + Deliberately the same shape of computation as + :func:`analytics.factor.forward_returns` (restrict to the evaluation grid, mask + a non-positive denominator to NaN, then ``groupby(symbol).shift(-h)``), so the + ``exec_to_exec`` series and the ``close_to_close`` series it replaces are + aligned by exactly the same rule and differ ONLY in the price they are built + from. ``h`` counts EVALUATION PERIODS: a date the grid does not contain is not + a next period, no matter what the minute cache holds for it. + """ + if not isinstance(horizon, int) or isinstance(horizon, bool) or horizon <= 0: + raise ValueError( + f"exec_forward_returns: horizon must be a positive int (evaluation " + f"periods); got {horizon!r}." + ) + on_grid = adjusted_price[ + adjusted_price.index.get_level_values(DATE_LEVEL).isin(dates) + ].sort_index() + safe = on_grid.where(on_grid > 0.0) + grouped = safe.groupby(level=SYMBOL_LEVEL, sort=False, group_keys=False) + forward = grouped.shift(-horizon) + return (forward / safe - 1.0).rename(f"forward_return_{horizon}d") + + +# --------------------------------------------------------------------------- # +# Coverage loss vs the close_to_close basis (the price of the new basis) +# --------------------------------------------------------------------------- # +def coverage_loss( + exec_returns: pd.Series, + close_returns: pd.Series, + status: pd.Series, + horizon: int, +) -> dict[str, object]: + """What the exec basis costs, in (date, symbol) pairs, BY CAUSE. + + A pair is "lost" when ``close_to_close`` produced a finite return and + ``exec_to_exec`` did not. The cause is read off the pair's ENTRY anchor status + first and its EXIT anchor status otherwise — the entry is what a trade would + have hit first, and attributing to one cause keeps the counts a partition + rather than an overlapping tally. + """ + close = close_returns.reindex(exec_returns.index) + lost_mask = np.isfinite(close.to_numpy(dtype=float)) & ~np.isfinite( + exec_returns.to_numpy(dtype=float) + ) + measurable = int(np.isfinite(close.to_numpy(dtype=float)).sum()) + lost_index = exec_returns.index[lost_mask] + + entry_status = status.reindex(exec_returns.index) + # The exit anchor is h evaluation periods later WITHIN the symbol — the same + # positional shift the returns used, so causes line up with the returns. + exit_status = entry_status.groupby( + level=SYMBOL_LEVEL, sort=False, group_keys=False + ).shift(-horizon) + # Entry first, exit otherwise: one cause per lost pair, so the counts partition + # the loss instead of double-counting a pair blocked at both ends. + cause = entry_status.where(entry_status != STATUS_OK).fillna(exit_status) + lost_cause = cause[lost_mask] + counts = lost_cause.value_counts() + by_cause = {r: int(counts.get(r, 0)) for r in MISS_REASONS} + # Neither anchor names a cause (both read "ok"): only reachable via a + # non-positive price. Counted apart rather than folded into a cause it does + # not belong to. + unattributed = int(len(lost_cause) - sum(by_cause.values())) + lost_symbols = set(lost_index.get_level_values(SYMBOL_LEVEL).astype(str)) + return { + "close_to_close_measurable_pairs": measurable, + "exec_to_exec_measurable_pairs": int( + np.isfinite(exec_returns.to_numpy(dtype=float)).sum() + ), + "lost_pairs": int(lost_mask.sum()), + "lost_pairs_pct_of_close_to_close": ( + float(lost_mask.sum()) / measurable * 100.0 if measurable else float("nan") + ), + "lost_pairs_by_cause": by_cause, + "lost_pairs_unattributed": unattributed, + "distinct_symbols_affected": len(lost_symbols), + } + + +__all__ = [ + "ARTIFACT_COLUMNS", + "ARTIFACT_SCHEMA_VERSION", + "MISS_BAD_ADJ_FACTOR", + "MISS_BAD_VWAP", + "MISS_NO_BAR", + "MISS_REASONS", + "STATUS_OK", + "ExecBasisParams", + "ExecPricePanel", + "artifact_key", + "build_exec_price_panel", + "coverage_loss", + "exec_forward_returns", + "intraday_spec_variant", +] diff --git a/tests/test_exec_basis_eval.py b/tests/test_exec_basis_eval.py new file mode 100644 index 0000000..b88d4e4 --- /dev/null +++ b/tests/test_exec_basis_eval.py @@ -0,0 +1,269 @@ +"""The runner-facing exec-basis evaluation glue — network-free, end to end. + +Checks the three things the eleven runners depend on and that a unit test of the +return builder alone would not catch: + + * the exec reports are written UNDER THEIR OWN names, so the accepted + close_to_close artifacts cannot be overwritten; + * the mandatory disclosure (execution parameters, coverage loss by cause, + measured live-call count) actually reaches the rendered report; + * adding that disclosure does NOT move a verdict axis — the Tradable axis stays + NOT_ASSESSED, because a return basis is not a fill-feasibility measurement. +""" + +from __future__ import annotations + +import json +import logging + +import numpy as np +import pandas as pd + +from analytics.eval.config import EvalConfig +from analytics.eval.verdict import AXIS_NOT_ASSESSED +from data.cache.intraday_cache import ENDPOINT as MINUTE_ENDPOINT +from data.cache.intraday_parquet_store import KEY_COLS, IntradayParquetStore +from data.clean.intraday_schema import RAW_INTRADAY_FREQ +from factors.spec import FactorSpec +from qt.config import ( + AlphaCfg, + BacktestCfg, + CacheCfg, + CostCfg, + DataCfg, + FactorCfg, + OutputCfg, + PortfolioCfg, + RootConfig, + UniverseCfg, +) +from qt.exec_basis_eval import run_exec_basis_evaluation + +LOGGER = logging.getLogger("test.exec_basis_eval") +SYMBOLS = ("A.SZ", "B.SZ", "C.SZ", "D.SZ") + + +def _days(n: int) -> list[pd.Timestamp]: + """``n`` weekday sessions — a daily grid the evaluator accepts as daily.""" + out: list[pd.Timestamp] = [] + day = pd.Timestamp("2024-01-01") + while len(out) < n: + if day.weekday() < 5: + out.append(day) + day += pd.Timedelta(days=1) + return out + + +def _minute_rows(symbol: str, day: pd.Timestamp, amount: float) -> list[dict]: + rows = [] + for offset, (close, vol, amt) in enumerate( + [(9.0, 10.0, 90.0), (9.5, 10.0, 95.0), (10.0, 100.0, amount), (11.0, 10.0, 110.0)] + ): + end = day + pd.Timedelta("14:49:00") + pd.Timedelta(minutes=offset) + rows.append( + { + "symbol": symbol, "bar_end": end, "source_trade_time": end, + "open": close, "high": close, "low": close, "close": close, + "volume": vol, "amount": amt, "freq": RAW_INTRADAY_FREQ, + } + ) + return rows + + +def _fixture(tmp_path, n_days: int = 40): + days = _days(n_days) + rng = np.random.default_rng(7) + root = tmp_path / "cache" + store = IntradayParquetStore(str(root)) + + price = {s: 1000.0 for s in SYMBOLS} + rows: list[dict] = [] + closes: dict[tuple[pd.Timestamp, str], float] = {} + for day in days: + for sym in SYMBOLS: + price[sym] *= 1.0 + float(rng.normal(0.0, 0.02)) + closes[(day, sym)] = price[sym] + # exec bar VWAP == amount / volume == price (volume 100). + rows.extend(_minute_rows(sym, day, price[sym] * 100.0)) + frame = pd.DataFrame(rows) + for sym, part in frame.groupby("symbol"): + store.upsert(MINUTE_ENDPOINT, str(sym), RAW_INTRADAY_FREQ, part, KEY_COLS) + + index = pd.MultiIndex.from_product( + [pd.DatetimeIndex(days), list(SYMBOLS)], names=["date", "symbol"] + ) + close_values = [closes[(d, s)] for d, s in index] + panel = pd.DataFrame( + { + "open": close_values, "high": close_values, "low": close_values, + "close": close_values, "volume": 100.0, "amount": 100.0, + "adj_factor": 1.0, + }, + index=index, + ) + factor = pd.Series( + rng.normal(size=len(index)), index=index, name="demo_minute_factor" + ) + book = pd.DataFrame({"book_a": rng.normal(size=len(index))}, index=index) + cfg = RootConfig( + data=DataCfg( + source="tushare", + start=str(days[0].date()), + end=str(days[-1].date()), + external_secret_file="/nonexistent.json", + cache=CacheCfg(enabled=True, root_dir=str(root)), + ), + universe=UniverseCfg(type="static", symbols=list(SYMBOLS)), + factors=[FactorCfg(name="momentum_20")], + alpha=AlphaCfg(), portfolio=PortfolioCfg(top_n=2), + backtest=BacktestCfg(), cost=CostCfg(), + output=OutputCfg( + data_dir=str(tmp_path / "data"), report_dir=str(tmp_path / "reports") + ), + ) + spec = FactorSpec( + factor_id="demo_minute_factor", version="1.0", + description="synthetic minute-derived factor", expected_ic_sign=1, + is_intraday=False, forward_return_horizon=1, + return_basis="close_to_close", input_fields=("close",), + family="microstructure", + ) + eval_cfg = EvalConfig( + universe="static", universe_is_pit=False, + start=str(days[0].date()), end=str(days[-1].date()), + is_exploratory=True, post_hoc_selected=False, rebalance="daily", + ) + return cfg, panel, factor, book, spec, eval_cfg, tmp_path / "reports" + + +def test_exec_basis_eval_writes_its_own_reports_and_leaves_the_control_alone(tmp_path): + cfg, panel, factor, book, spec, eval_cfg, report_dir = _fixture(tmp_path) + report_dir.mkdir(parents=True, exist_ok=True) + # Stand-ins for the accepted close_to_close artifacts. + control = { + report_dir / "demo_no_book.md": "CLOSE-TO-CLOSE CONTROL", + report_dir / "demo_with_book.md": "CLOSE-TO-CLOSE CONTROL", + } + for path, text in control.items(): + path.write_text(text, encoding="utf-8") + + out = run_exec_basis_evaluation( + factor, spec, eval_cfg, book, + cfg=cfg, panel=panel, symbols=list(SYMBOLS), logger=LOGGER, + report_dir=report_dir, stem="demo", + ) + + for path, text in control.items(): + assert path.read_text(encoding="utf-8") == text # untouched + assert out.no_book_md.name == "demo_exec_no_book.md" + assert out.with_book_md.name == "demo_exec_with_book.md" + for path in (out.no_book_md, out.with_book_md, out.no_book_json, + out.with_book_json, out.sanity_report_path, + out.no_book_dashboard, out.with_book_dashboard): + assert path.exists() + assert out.spec.is_intraday is True + assert out.spec.return_basis == "exec_to_exec" + assert out.minute_live_calls == 0 + + +def test_exec_basis_eval_discloses_parameters_and_coverage_in_every_report(tmp_path): + cfg, panel, factor, book, spec, eval_cfg, report_dir = _fixture(tmp_path) + out = run_exec_basis_evaluation( + factor, spec, eval_cfg, book, + cfg=cfg, panel=panel, symbols=list(SYMBOLS), logger=LOGGER, + report_dir=report_dir, stem="demo", + ) + for path in (out.no_book_md, out.with_book_md): + text = path.read_text(encoding="utf-8") + assert "exec_to_exec" in text + assert "14:51:00" in text # execution window + assert "bar_vwap" in text # price basis + assert "adj_factor" in text # the adjustment identity + assert "stk_mins_live_calls" in text + for cause in ("no_bar", "bad_vwap", "bad_adj_factor"): + assert f"lost_pairs_by_cause_{cause}" in text + # The bad_vwap loss is small but DIRECTIONAL — the excluded minutes are + # exactly the ones with no traded volume. The percentage must never stand + # alone as if smallness made it random. + assert "LIQUIDITY-CORRELATED, NON-RANDOM" in text + sanity_text = out.sanity_report_path.read_text(encoding="utf-8") + assert "liquidity-correlated, non-random" in sanity_text + for path in (out.no_book_json, out.with_book_json): + payload = json.loads(path.read_text(encoding="utf-8")) + blob = json.dumps(payload) + assert "exec_to_exec" in blob + assert "sanity_corr_vs_close_to_close_median" in blob + # The forward-return provenance says the returns came from OUTSIDE the evaluator. + assert "computed OUTSIDE this evaluator" in out.no_book_md.read_text(encoding="utf-8") + + +def test_exec_basis_disclosure_does_not_move_the_tradable_axis(tmp_path): + """A return-basis disclosure must not read as measured fill feasibility.""" + cfg, panel, factor, book, spec, eval_cfg, report_dir = _fixture(tmp_path) + out = run_exec_basis_evaluation( + factor, spec, eval_cfg, book, + cfg=cfg, panel=panel, symbols=list(SYMBOLS), logger=LOGGER, + report_dir=report_dir, stem="demo", + ) + for report in (out.no_book, out.with_book): + assert report.require_verdict().tradable.verdict == AXIS_NOT_ASSESSED + assert out.no_book_metrics["tradable"] == AXIS_NOT_ASSESSED + + +def test_exec_basis_eval_shares_one_artifact_across_factors(tmp_path): + """Two factors on the same universe/window/parameters get the SAME returns. + + This is what makes cross-factor comparison meaningful: eleven factors scored + against eleven slightly different return series would not be comparable. + """ + cfg, panel, factor, book, spec, eval_cfg, report_dir = _fixture(tmp_path) + first = run_exec_basis_evaluation( + factor, spec, eval_cfg, book, + cfg=cfg, panel=panel, symbols=list(SYMBOLS), logger=LOGGER, + report_dir=report_dir, stem="demo_one", + ) + other_spec = FactorSpec( + factor_id="other_minute_factor", version="2.0", + description="a different synthetic minute factor", expected_ic_sign=-1, + is_intraday=False, forward_return_horizon=1, + return_basis="close_to_close", input_fields=("close",), + ) + other_factor = factor.rename("other_minute_factor") * -1.0 + second = run_exec_basis_evaluation( + other_factor, other_spec, eval_cfg, book, + cfg=cfg, panel=panel, symbols=list(SYMBOLS), logger=LOGGER, + report_dir=report_dir, stem="demo_two", + ) + assert first.artifact_key == second.artifact_key + assert first.artifact_reused is False + assert second.artifact_reused is True + assert first.coverage == second.coverage + + +def test_exec_basis_cli_line_survives_an_absent_metric(tmp_path): + """A Skipped section leaves None metrics; the summary must not raise on them. + + The print runs after four reports have been written and outside the command's + error handling, so a TypeError here would bury finished work in a traceback. + """ + cfg, panel, factor, book, spec, eval_cfg, report_dir = _fixture(tmp_path, n_days=30) + out = run_exec_basis_evaluation( + factor, spec, eval_cfg, book, + cfg=cfg, panel=panel, symbols=list(SYMBOLS), logger=LOGGER, + report_dir=report_dir, stem="demo", + ) + import dataclasses + + from qt.exec_basis_eval import format_exec_basis_line + + blanked = dataclasses.replace( + out, + no_book_metrics={**out.no_book_metrics, "ic_mean": None, "ic_ir": None}, + with_book_metrics={**out.with_book_metrics, "incremental_ic_ir": None}, + ) + line = format_exec_basis_line(blanked) + assert "ic_mean=n/a" in line + assert "incr_ic_ir=n/a" in line + # the measured facts are still reported + assert "stk_mins_live_calls=0" in line + assert "no_bar=" in line diff --git a/tests/test_exec_forward_returns.py b/tests/test_exec_forward_returns.py new file mode 100644 index 0000000..12dd271 --- /dev/null +++ b/tests/test_exec_forward_returns.py @@ -0,0 +1,572 @@ +"""exec-to-exec forward returns: pricing, alignment, adjustment and missingness. + +Every test here is written to FAIL under a specific plausible mistake, and the four +mutations named in the task card (§4) are checked against these tests by actually +patching the implementation and observing the failure — not by assertion in prose. + +Naming note: test function names must be unique across the whole suite (the test +package has no ``__init__.py``, so two identically named functions in different +files silently collapse into one). Everything below is prefixed accordingly. +""" + +from __future__ import annotations + +import logging + +import numpy as np +import pandas as pd +import pytest + +from analytics.eval.config import EvalConfig +from analytics.eval.ir import EvalContext, build_eval_ir +from data.cache.intraday_cache import ENDPOINT as MINUTE_ENDPOINT +from data.cache.intraday_parquet_store import KEY_COLS, IntradayParquetStore +from data.clean.intraday_schema import RAW_INTRADAY_FREQ, normalize_intraday_bars +from factors.spec import FactorSpec +from qt.config import ( + AlphaCfg, + BacktestCfg, + CacheCfg, + CostCfg, + DataCfg, + FactorCfg, + IntradayCfg, + OutputCfg, + PortfolioCfg, + RootConfig, + UniverseCfg, +) +from qt.exec_basis_sanity import check_exec_basis +from qt.exec_forward_returns import ( + MISS_BAD_ADJ_FACTOR, + MISS_BAD_VWAP, + MISS_NO_BAR, + STATUS_OK, + ExecBasisParams, + _restrict_to_execution_window, + artifact_key, + build_exec_price_panel, + coverage_loss, + exec_forward_returns, + intraday_spec_variant, +) +from runtime.intraday_execution import build_execution_prices + +LOGGER = logging.getLogger("test.exec_basis") + +DAYS = ("2024-01-02", "2024-01-03", "2024-01-04", "2024-01-05") + + +# --------------------------------------------------------------------------- # +# Fixtures: a tiny minute cache + a tiny daily panel +# --------------------------------------------------------------------------- # +def _bar(symbol: str, day: str, hhmmss: str, close: float, volume: float, amount: float) -> dict: + end = pd.Timestamp(day) + pd.Timedelta(hhmmss) + return { + "symbol": symbol, + "bar_end": end, + "source_trade_time": end, + "open": close, + "high": close, + "low": close, + "close": close, + "volume": volume, + "amount": amount, + "freq": RAW_INTRADAY_FREQ, + } + + +def _session(symbol: str, day: str, *, close: float, volume: float, amount: float) -> list[dict]: + """One session whose 14:51 bar has a VWAP DIFFERENT from its close. + + 14:49 / 14:50 bars sit before the execution window and 14:52 after the earliest + in-window bar, so a run that picked the wrong bar shows up as a wrong number + rather than a wrong-looking shape. + """ + return [ + _bar(symbol, day, "14:49:00", close * 3.0, 10.0, close * 30.0), + _bar(symbol, day, "14:50:00", close * 2.0, 10.0, close * 20.0), + _bar(symbol, day, "14:51:00", close, volume, amount), + _bar(symbol, day, "14:52:00", close * 5.0, 10.0, close * 50.0), + ] + + +def _write_cache(root, rows: list[dict]) -> IntradayParquetStore: + store = IntradayParquetStore(str(root)) + frame = pd.DataFrame(rows) + for symbol, part in frame.groupby("symbol"): + store.upsert(MINUTE_ENDPOINT, str(symbol), RAW_INTRADAY_FREQ, part, KEY_COLS) + return store + + +def _panel(symbols: tuple[str, ...], days: tuple[str, ...], adj: dict | None = None) -> pd.DataFrame: + """Daily panel: only ``adj_factor`` and the (date, symbol) grid matter here.""" + index = pd.MultiIndex.from_product( + [pd.DatetimeIndex([pd.Timestamp(d) for d in days]), list(symbols)], + names=["date", "symbol"], + ) + factors = [ + (adj or {}).get((pd.Timestamp(d).strftime("%Y-%m-%d"), s), 1.0) + for d, s in index + ] + return pd.DataFrame( + { + "open": 10.0, + "high": 10.0, + "low": 10.0, + "close": 10.0, + "volume": 1.0, + "amount": 10.0, + "adj_factor": factors, + }, + index=index, + ) + + +def _config(root, start: str, end: str, symbols: tuple[str, ...], **intraday) -> RootConfig: + return RootConfig( + data=DataCfg( + source="tushare", + start=start, + end=end, + external_secret_file="/nonexistent.json", + cache=CacheCfg(enabled=True, root_dir=str(root)), + ), + universe=UniverseCfg(type="static", symbols=list(symbols)), + factors=[FactorCfg(name="momentum_20")], + alpha=AlphaCfg(), + portfolio=PortfolioCfg(top_n=1), + backtest=BacktestCfg(), + cost=CostCfg(), + output=OutputCfg(data_dir=str(root.parent / "data")), + intraday=IntradayCfg(enabled=True, **intraday) if intraday else None, + ) + + +def _build(tmp_path, rows, symbols, days, adj=None, cfg=None): + root = tmp_path / "cache" + _write_cache(root, rows) + panel = _panel(symbols, days, adj) + conf = cfg or _config(root, days[0], days[-1], symbols) + params = ExecBasisParams.from_config(conf) + prices = build_exec_price_panel(conf, panel, list(symbols), params, LOGGER) + return conf, panel, params, prices + + +# --------------------------------------------------------------------------- # +# 1. Pricing: the VWAP of the selected bar, never its close +# --------------------------------------------------------------------------- # +def test_exec_basis_price_is_bar_vwap_never_bar_close(tmp_path): + # 14:51 bar: close 10.0, volume 100, amount 1200 -> VWAP 12.0 != close. + rows = _session("AAA.SZ", DAYS[0], close=10.0, volume=100.0, amount=1200.0) + _, _, _, prices = _build(tmp_path, rows, ("AAA.SZ",), (DAYS[0],)) + + row = prices.frame.loc[(pd.Timestamp(DAYS[0]), "AAA.SZ")] + assert row["status"] == STATUS_OK + assert row["raw_exec_price"] == pytest.approx(12.0) # amount / volume + assert row["raw_exec_price"] != pytest.approx(10.0) # NOT the bar close + assert row["exec_time"] == pd.Timestamp(DAYS[0]) + pd.Timedelta("14:51:00") + + +def test_exec_basis_selects_the_earliest_in_window_bar(tmp_path): + """A 14:52 bar must never win over an existing 14:51 bar.""" + rows = _session("AAA.SZ", DAYS[0], close=10.0, volume=100.0, amount=1200.0) + _, _, _, prices = _build(tmp_path, rows, ("AAA.SZ",), (DAYS[0],)) + # the 14:52 bar's VWAP would be 50.0/10 = 5.0 * ... -> a very different number + assert prices.frame.loc[(pd.Timestamp(DAYS[0]), "AAA.SZ"), "raw_exec_price"] == pytest.approx(12.0) + + +# --------------------------------------------------------------------------- # +# 2. Adjustment: the ratio identity, and its invariances +# --------------------------------------------------------------------------- # +def test_exec_basis_return_is_the_adjusted_price_ratio(tmp_path): + rows = ( + _session("AAA.SZ", DAYS[0], close=10.0, volume=100.0, amount=1000.0) + + _session("AAA.SZ", DAYS[1], close=11.0, volume=100.0, amount=1100.0) + ) + adj = {(DAYS[0], "AAA.SZ"): 2.0, (DAYS[1], "AAA.SZ"): 2.0} + _, panel, _, prices = _build(tmp_path, rows, ("AAA.SZ",), DAYS[:2], adj) + + dates = pd.DatetimeIndex([pd.Timestamp(d) for d in DAYS[:2]], name="date") + returns = exec_forward_returns(prices.adjusted_price, dates, 1) + got = returns.loc[(pd.Timestamp(DAYS[0]), "AAA.SZ")] + assert got == pytest.approx((11.0 * 2.0) / (10.0 * 2.0) - 1.0) + + +def test_exec_basis_adjustment_is_invariant_to_a_common_rescale(tmp_path): + """Scaling every adj_factor by a constant cannot move a single return. + + This is the property that makes the identity exact and window-invariant (the + per-symbol anchor cancels). A return that moved would mean the adjustment was + applied as something other than a ratio. + """ + rows = ( + _session("AAA.SZ", DAYS[0], close=10.0, volume=100.0, amount=1000.0) + + _session("AAA.SZ", DAYS[1], close=11.0, volume=100.0, amount=1210.0) + ) + dates = pd.DatetimeIndex([pd.Timestamp(d) for d in DAYS[:2]], name="date") + base = {(DAYS[0], "AAA.SZ"): 3.0, (DAYS[1], "AAA.SZ"): 1.5} + scaled = {k: v * 7.0 for k, v in base.items()} + + _, _, _, p1 = _build(tmp_path / "a", rows, ("AAA.SZ",), DAYS[:2], base) + _, _, _, p2 = _build(tmp_path / "b", rows, ("AAA.SZ",), DAYS[:2], scaled) + r1 = exec_forward_returns(p1.adjusted_price, dates, 1) + r2 = exec_forward_returns(p2.adjusted_price, dates, 1) + pd.testing.assert_series_equal(r1, r2) + + +def test_exec_basis_ex_date_period_is_corporate_action_free(tmp_path): + """A 50% ex-date drop in the RAW price must not read as a -50% return.""" + rows = ( + _session("AAA.SZ", DAYS[0], close=20.0, volume=100.0, amount=2000.0) + + _session("AAA.SZ", DAYS[1], close=10.0, volume=100.0, amount=1010.0) + ) + # adj_factor doubles across the ex-date, exactly offsetting the price halving. + adj = {(DAYS[0], "AAA.SZ"): 1.0, (DAYS[1], "AAA.SZ"): 2.0} + _, _, _, prices = _build(tmp_path, rows, ("AAA.SZ",), DAYS[:2], adj) + dates = pd.DatetimeIndex([pd.Timestamp(d) for d in DAYS[:2]], name="date") + got = exec_forward_returns(prices.adjusted_price, dates, 1).loc[ + (pd.Timestamp(DAYS[0]), "AAA.SZ") + ] + unadjusted = 10.1 / 20.0 - 1.0 + assert unadjusted == pytest.approx(-0.495) # what a raw ratio would report + assert got == pytest.approx((10.1 * 2.0) / (20.0 * 1.0) - 1.0) + assert got == pytest.approx(0.01) # the real +1% move survives + + +# --------------------------------------------------------------------------- # +# 3. Alignment: h steps on the EVALUATION grid +# --------------------------------------------------------------------------- # +def test_exec_basis_steps_the_evaluation_grid_not_the_data_grid(tmp_path): + """The exit anchor is the next EVALUATION period, not the next cached day. + + The minute cache and the daily panel both hold three sessions; the factor is + only evaluated on the first and third. The forward return at D1 must therefore + be measured against D3 — a series that reached for D2 (the next day the data + happens to have) is measuring a different holding period entirely. + """ + rows = ( + _session("AAA.SZ", DAYS[0], close=10.0, volume=100.0, amount=1000.0) + + _session("AAA.SZ", DAYS[1], close=10.0, volume=100.0, amount=5000.0) + + _session("AAA.SZ", DAYS[2], close=10.0, volume=100.0, amount=2000.0) + ) + _, _, _, prices = _build(tmp_path, rows, ("AAA.SZ",), DAYS[:3]) + + evaluation_grid = pd.DatetimeIndex( + [pd.Timestamp(DAYS[0]), pd.Timestamp(DAYS[2])], name="date" + ) + returns = exec_forward_returns(prices.adjusted_price, evaluation_grid, 1) + got = returns.loc[(pd.Timestamp(DAYS[0]), "AAA.SZ")] + assert got == pytest.approx(20.0 / 10.0 - 1.0) # D1 -> D3 (evaluation grid) + assert got != pytest.approx(50.0 / 10.0 - 1.0) # NOT D1 -> D2 (data grid) + # D2 is not on the evaluation grid at all, so it carries no return. + assert (pd.Timestamp(DAYS[1]), "AAA.SZ") not in returns.index + + +def test_exec_basis_horizon_counts_periods_not_days(tmp_path): + rows = [ + bar + for i, day in enumerate(DAYS[:3]) + for bar in _session("AAA.SZ", day, close=10.0, volume=100.0, amount=1000.0 * (i + 1)) + ] + _, _, _, prices = _build(tmp_path, rows, ("AAA.SZ",), DAYS[:3]) + dates = pd.DatetimeIndex([pd.Timestamp(d) for d in DAYS[:3]], name="date") + two = exec_forward_returns(prices.adjusted_price, dates, 2) + assert two.loc[(pd.Timestamp(DAYS[0]), "AAA.SZ")] == pytest.approx(30.0 / 10.0 - 1.0) + assert np.isnan(two.loc[(pd.Timestamp(DAYS[1]), "AAA.SZ")]) + + +# --------------------------------------------------------------------------- # +# 4. Missingness: three causes, counted apart, never softened +# --------------------------------------------------------------------------- # +def test_exec_basis_missing_bar_is_no_bar_and_never_a_close_fallback(tmp_path): + """A session with bars only OUTSIDE the execution window prices at nothing.""" + rows = [ + _bar("AAA.SZ", DAYS[0], "14:49:00", 10.0, 100.0, 1000.0), + _bar("AAA.SZ", DAYS[0], "14:50:00", 10.0, 100.0, 1000.0), + ] + _, _, _, prices = _build(tmp_path, rows, ("AAA.SZ",), (DAYS[0],)) + row = prices.frame.loc[(pd.Timestamp(DAYS[0]), "AAA.SZ")] + assert row["status"] == MISS_NO_BAR + assert np.isnan(row["raw_exec_price"]) + assert np.isnan(row["adj_exec_price"]) + assert prices.status_counts()[MISS_NO_BAR] == 1 + + +@pytest.mark.parametrize( + "volume, amount", + [(0.0, 1000.0), (100.0, 0.0), (float("nan"), 1000.0), (100.0, float("nan"))], +) +def test_exec_basis_undefined_vwap_is_bad_vwap_never_the_bar_close(tmp_path, volume, amount): + """No traded shares (or value) means no traded average price — not the close.""" + rows = _session("AAA.SZ", DAYS[0], close=10.0, volume=volume, amount=amount) + _, _, _, prices = _build(tmp_path, rows, ("AAA.SZ",), (DAYS[0],)) + row = prices.frame.loc[(pd.Timestamp(DAYS[0]), "AAA.SZ")] + assert row["status"] == MISS_BAD_VWAP + assert np.isnan(row["raw_exec_price"]) + assert not np.isclose(np.nan_to_num(row["raw_exec_price"], nan=-1.0), 10.0) + assert prices.status_counts()[MISS_BAD_VWAP] == 1 + assert prices.status_counts()[MISS_NO_BAR] == 0 + + +@pytest.mark.parametrize("factor", [float("nan"), 0.0, -1.0]) +def test_exec_basis_unusable_adj_factor_is_counted_never_assumed_one(tmp_path, factor): + """A missing/non-positive adj_factor blocks the pair; 1.0 is NOT substituted.""" + rows = _session("AAA.SZ", DAYS[0], close=10.0, volume=100.0, amount=1200.0) + adj = {(DAYS[0], "AAA.SZ"): factor} + _, _, _, prices = _build(tmp_path, rows, ("AAA.SZ",), (DAYS[0],), adj) + row = prices.frame.loc[(pd.Timestamp(DAYS[0]), "AAA.SZ")] + assert row["status"] == MISS_BAD_ADJ_FACTOR + assert row["raw_exec_price"] == pytest.approx(12.0) # the bar priced fine ... + assert np.isnan(row["adj_exec_price"]) # ... the pair still blocks + assert prices.status_counts()[MISS_BAD_ADJ_FACTOR] == 1 + + +def test_exec_basis_uncached_symbol_is_no_bar_without_any_live_call(tmp_path): + rows = _session("AAA.SZ", DAYS[0], close=10.0, volume=100.0, amount=1200.0) + _, _, _, prices = _build(tmp_path, rows, ("AAA.SZ", "ZZZ.SZ"), (DAYS[0],)) + assert prices.minute_live_calls == 0 + assert prices.symbols_with_bars == 1 + assert prices.frame.loc[(pd.Timestamp(DAYS[0]), "ZZZ.SZ"), "status"] == MISS_NO_BAR + + +def test_exec_basis_coverage_loss_is_partitioned_by_cause(): + index = pd.MultiIndex.from_tuples( + [ + (pd.Timestamp(DAYS[0]), "A.SZ"), + (pd.Timestamp(DAYS[0]), "B.SZ"), + (pd.Timestamp(DAYS[0]), "C.SZ"), + (pd.Timestamp(DAYS[0]), "D.SZ"), + ], + names=["date", "symbol"], + ) + exec_returns = pd.Series([0.01, np.nan, np.nan, np.nan], index=index) + close_returns = pd.Series([0.01, 0.02, 0.03, np.nan], index=index) + status = pd.Series([STATUS_OK, MISS_NO_BAR, MISS_BAD_VWAP, MISS_BAD_ADJ_FACTOR], index=index) + + out = coverage_loss(exec_returns, close_returns, status, 1) + assert out["lost_pairs"] == 2 # D has no close return -> not a loss + assert out["lost_pairs_by_cause"] == { + MISS_NO_BAR: 1, + MISS_BAD_VWAP: 1, + MISS_BAD_ADJ_FACTOR: 0, + } + assert out["distinct_symbols_affected"] == 2 + assert out["close_to_close_measurable_pairs"] == 3 + assert out["exec_to_exec_measurable_pairs"] == 1 + + +# --------------------------------------------------------------------------- # +# 5. Reuse of the canonical execution layer +# --------------------------------------------------------------------------- # +def test_exec_basis_window_prefilter_matches_a_full_day_build(tmp_path): + """The performance pre-filter must be semantically invisible. + + Building from the whole session and building from only the in-window bars must + produce the same prices and the same fills — otherwise the pre-filter, not + ``resolve_fill``, would be deciding which bar executes. + """ + rows = ( + _session("AAA.SZ", DAYS[0], close=10.0, volume=100.0, amount=1200.0) + + _session("AAA.SZ", DAYS[1], close=11.0, volume=100.0, amount=1300.0) + ) + stored = pd.DataFrame(rows) + params = ExecBasisParams.from_config( + _config(tmp_path / "cache", DAYS[0], DAYS[1], ("AAA.SZ",)) + ) + cfg = params.exec_config() + dates = [pd.Timestamp(DAYS[0]), pd.Timestamp(DAYS[1])] + + def _norm(frame): + return normalize_intraday_bars( + frame.rename(columns={"bar_end": "time"})[ + ["time", "symbol", "open", "high", "low", "close", "volume", + "amount", "source_trade_time"] + ], + freq=RAW_INTRADAY_FREQ, + data_lag=params.data_lag, + ) + + full_prices, full_fills = build_execution_prices(_norm(stored), dates, ["AAA.SZ"], cfg) + win_prices, win_fills = build_execution_prices( + _norm(_restrict_to_execution_window(stored, params)), dates, ["AAA.SZ"], cfg + ) + pd.testing.assert_frame_equal(full_prices, win_prices) + assert full_fills == win_fills + + +# --------------------------------------------------------------------------- # +# 6. The spec variant and the evaluator's guard +# --------------------------------------------------------------------------- # +def _daily_spec() -> FactorSpec: + return FactorSpec( + factor_id="demo_minute_factor", + version="1.0", + description="synthetic minute-derived factor", + expected_ic_sign=1, + is_intraday=False, + forward_return_horizon=1, + return_basis="close_to_close", + input_fields=("close",), + family="microstructure", + ) + + +def test_exec_basis_spec_variant_carries_the_execution_parameters(tmp_path): + cfg = _config( + tmp_path / "cache", DAYS[0], DAYS[1], ("AAA.SZ",), + decision_time="14:45:00", execution_window=("14:46:00", "14:50:00"), + ) + params = ExecBasisParams.from_config(cfg) + variant = intraday_spec_variant(_daily_spec(), params) + + assert variant.is_intraday is True + assert variant.return_basis == "exec_to_exec" + assert variant.decision_cutoff == "14:45:00" + assert variant.execution_window == "[14:46:00,14:50:00]" + assert variant.execution_model == params.execution_model + assert variant.session_open == params.session_open + assert variant.data_lag == params.data_lag + # The FACTOR is untouched — only what it is scored against changed. + daily = _daily_spec() + assert (variant.factor_id, variant.version, variant.expected_ic_sign) == ( + daily.factor_id, daily.version, daily.expected_ic_sign + ) + # And the declared parameters are the ones the returns were computed with. + exec_cfg = params.exec_config() + assert variant.decision_cutoff == exec_cfg.decision_time + assert variant.execution_window == f"[{exec_cfg.execution_window[0]},{exec_cfg.execution_window[1]}]" + + +@pytest.mark.parametrize( + "field", ["decision_cutoff", "data_lag", "session_open", "execution_model", "execution_window"] +) +def test_exec_basis_spec_variant_needs_the_whole_minute_block(field): + """Dropping any one of the five must fail at FactorSpec construction.""" + import dataclasses + + params = ExecBasisParams.from_config( + RootConfig( + data=DataCfg(source="demo", start="2024-01-02", end="2024-01-05"), + universe=UniverseCfg(type="static", symbols=["A.SZ"]), + factors=[FactorCfg(name="momentum_20")], + alpha=AlphaCfg(), portfolio=PortfolioCfg(top_n=1), + backtest=BacktestCfg(), cost=CostCfg(), output=OutputCfg(), + ) + ) + block = params.spec_fields() + block[field] = None + with pytest.raises(ValueError, match="minute block"): + dataclasses.replace( + _daily_spec(), is_intraday=True, return_basis="exec_to_exec", **block + ) + + +def test_exec_basis_evaluator_still_refuses_exec_basis_without_forward_returns(tmp_path): + """The frozen guard must NOT be weakened: no returns supplied -> raise. + + Without this the evaluator would silently score an ``exec_to_exec`` factor on + close-to-close returns and label the report ``exec_to_exec``. + """ + params = ExecBasisParams.from_config( + _config(tmp_path / "cache", DAYS[0], DAYS[1], ("AAA.SZ",)) + ) + spec = intraday_spec_variant(_daily_spec(), params) + index = pd.MultiIndex.from_product( + [pd.DatetimeIndex([pd.Timestamp(d) for d in DAYS]), ["A.SZ", "B.SZ"]], + names=["date", "symbol"], + ) + factor = pd.Series(np.arange(len(index), dtype=float), index=index) + panel = _panel(("A.SZ", "B.SZ"), DAYS) + cfg = EvalConfig( + universe="static", universe_is_pit=False, start=DAYS[0], end=DAYS[-1], + is_exploratory=True, post_hoc_selected=False, rebalance="daily", + ) + with pytest.raises(ValueError, match="cannot be derived from a close panel"): + build_eval_ir(factor, spec, cfg, EvalContext(price_panel=panel)) + + +# --------------------------------------------------------------------------- # +# 7. The shared artifact +# --------------------------------------------------------------------------- # +def test_exec_basis_artifact_is_reused_and_rekeyed_on_a_parameter_change(tmp_path): + rows = _session("AAA.SZ", DAYS[0], close=10.0, volume=100.0, amount=1200.0) + conf, panel, params, first = _build(tmp_path, rows, ("AAA.SZ",), (DAYS[0],)) + assert first.reused is False + assert first.path.exists() + + second = build_exec_price_panel(conf, panel, ["AAA.SZ"], params, LOGGER) + assert second.reused is True + assert second.path == first.path + pd.testing.assert_frame_equal(first.frame, second.frame) + + # A different execution window is a different return series -> different key. + other = ExecBasisParams.from_config( + _config(tmp_path / "cache", DAYS[0], DAYS[0], ("AAA.SZ",), + execution_window=("14:52:00", "14:56:59")) + ) + dates = pd.DatetimeIndex([pd.Timestamp(DAYS[0])], name="date") + assert artifact_key(conf, ["AAA.SZ"], dates, other)[0] != first.key + + +def test_exec_basis_params_default_to_the_project_minute_conventions(tmp_path): + """No intraday block -> the framework's declared defaults, and it says so.""" + conf = _config(tmp_path / "cache", DAYS[0], DAYS[1], ("AAA.SZ",)) + assert conf.intraday is None + params = ExecBasisParams.from_config(conf) + assert params.decision_cutoff == "14:50:00" + assert params.execution_window == ("14:51:00", "14:56:59") + assert params.execution_price_basis == "bar_vwap" + assert params.session_open == "09:30:00" + assert "defaults" in params.source + + +# --------------------------------------------------------------------------- # +# 8. The implementation-independent sanity checks +# --------------------------------------------------------------------------- # +def _sanity_fixture(tmp_path): + symbols = ("A.SZ", "B.SZ", "C.SZ") + amounts = {"A.SZ": (1000.0, 1010.0, 1030.0, 1020.0), + "B.SZ": (2000.0, 2040.0, 2010.0, 2050.0), + "C.SZ": (3000.0, 2970.0, 3030.0, 3060.0)} + rows = [ + bar + for sym in symbols + for i, day in enumerate(DAYS) + for bar in _session(sym, day, close=10.0, volume=100.0, amount=amounts[sym][i]) + ] + conf, panel, params, prices = _build(tmp_path, rows, symbols, DAYS) + dates = pd.DatetimeIndex([pd.Timestamp(d) for d in DAYS], name="date") + exec_returns = exec_forward_returns(prices.adjusted_price, dates, 1) + # A close panel that MOVES with the exec prices, so the agreement check is a + # real check rather than a comparison against a constant. + close = panel.copy() + close["close"] = prices.frame["raw_exec_price"].to_numpy() + from analytics.factor import forward_returns as _fwd + + close_returns = _fwd(close, periods=(1,))["forward_return_1d"] + return conf, params, prices, exec_returns, close_returns + + +def test_exec_basis_sanity_hand_check_agrees_with_the_generated_returns(tmp_path): + conf, params, prices, exec_returns, close_returns = _sanity_fixture(tmp_path) + sanity = check_exec_basis( + prices.frame, exec_returns, close_returns, params, + conf.data.cache.root_dir, 1, n_hand_checks=3, + ) + assert sanity.corr_ok + assert len(sanity.hand_checks) == 3 + assert sanity.hand_check_max_abs_diff < 1e-12 + for row in sanity.hand_checks: + assert row["entry_vwap"] == pytest.approx( + row["entry_amount"] / row["entry_volume"] + ) + + +def test_exec_basis_sanity_raises_when_agreement_collapses(tmp_path): + conf, params, prices, exec_returns, close_returns = _sanity_fixture(tmp_path) + with pytest.raises(ValueError, match="sanity check FAILED"): + check_exec_basis( + prices.frame, -exec_returns, close_returns, params, + conf.data.cache.root_dir, 1, n_hand_checks=1, + )