mirrors / stefan-jansen--machine-learning-for-trading

项目文件夹

stefan-jansen--machine-learning-for-trading
文件
main
T
wehub-resource-sync 0249a267e1 chore: import upstream snapshot with attribution
2026-07-13 13:26:28 +08:00

496 行
19 KiB
Python
原始文件 永久链接 Blame 文件历史

此文件含有模棱两可的 Unicode 字符
此文件含有可能会与其他字符混淆的 Unicode 字符。 如果您是想特意这样的，可以安全地忽略该警告。 使用 Escape 按钮显示他们。
"""Tests for case_studies/utils/analytics.py — registry query contracts.
Per memory rule ``feedback_results_json``: "Registry only, never JSONs."
This module is the canonical path Ch6–Ch20 insights notebooks take to pull
IC / AUC / Sharpe numbers into the book. A silent regression in any of
these queries would corrupt every cross-chapter summary table.
The tests pin three layers:
1. **Metadata invariants** — the handful of dicts that declare the 9
case-study IDs must agree on keys, so a new case study can't be added
to one dict and forgotten in another.
2. **Path resolution** — ``_cs_dir`` / ``_registry_path`` honor
``ML4T_OUTPUT_DIR`` for test isolation.
3. **Query contracts** — against a seeded SQLite registry:
- ``load_model_ic`` filters by family, split, case_studies list and
returns the expected rows with a ``case_study`` label column.
- ``load_classification_metrics`` requires ``task_type = 'classification'``.
- ``load_best_ic_per_family`` picks max IC per (case_study, family)
pair and optionally restricts to the primary label.
- ``load_chapter_backtests("ch16")`` maps to ``stage="signal"`` and
joins backtest_runs × backtest_metrics × prediction_sets ×
training_runs.
- Spec helpers: ``extract_cost_bps`` sums commission + slippage
from either v1 or v2 backtest specs; ``extract_allocator`` reads
strategy.allocation.method.
"""
from __future__ import annotations
import json
import sqlite3
from pathlib import Path
import polars as pl
import pytest
from case_studies.utils import analytics
# -----------------------------------------------------------------------------
# Metadata invariants
# -----------------------------------------------------------------------------
def test_case_study_ids_match_metadata_keys() -> None:
assert list(analytics.CASE_STUDY_META.keys()) == analytics.CASE_STUDY_IDS
@pytest.mark.parametrize(
"dict_name",
["PRIMARY_LABELS", "SHORT_NAMES", "DATASET_META", "CADENCE_MAP", "DISPLAY_NAMES"],
)
def test_metadata_dict_keys_match_case_study_ids(dict_name) -> None:
"""Every metadata dict must enumerate the same 9 case studies — otherwise
cross-dict joins in load_best_ic_per_family / load_chapter_backtests drop
rows silently.
"""
d = getattr(analytics, dict_name)
assert set(d.keys()) == set(analytics.CASE_STUDY_IDS), (
f"{dict_name} keys differ: missing {set(analytics.CASE_STUDY_IDS) - set(d.keys())}, "
f"extra {set(d.keys()) - set(analytics.CASE_STUDY_IDS)}"
)
def test_primary_labels_are_non_empty_strings() -> None:
for cs, lbl in analytics.PRIMARY_LABELS.items():
assert isinstance(lbl, str) and lbl, f"{cs}: empty/non-string primary label"
# -----------------------------------------------------------------------------
# Path resolution
# -----------------------------------------------------------------------------
def test_cs_dir_production_path(monkeypatch) -> None:
"""With no ML4T_OUTPUT_DIR, _cs_dir falls back to REPO_ROOT/case_studies."""
monkeypatch.delenv("ML4T_OUTPUT_DIR", raising=False)
from utils.paths import REPO_ROOT
assert analytics._cs_dir() == REPO_ROOT / "case_studies"
def test_cs_dir_redirects_to_output_dir_when_registry_present(tmp_path, monkeypatch) -> None:
"""With ML4T_OUTPUT_DIR set AND a registry.db present under tmp, _cs_dir
returns the tmp root instead of the production case_studies path.
"""
monkeypatch.setenv("ML4T_OUTPUT_DIR", str(tmp_path))
(tmp_path / "etfs" / "run_log").mkdir(parents=True)
(tmp_path / "etfs" / "run_log" / "registry.db").touch()
assert analytics._cs_dir("etfs") == tmp_path
def test_cs_dir_falls_back_when_registry_missing_under_output_dir(tmp_path, monkeypatch) -> None:
"""ML4T_OUTPUT_DIR set but no registry.db under it → fall back to production."""
monkeypatch.setenv("ML4T_OUTPUT_DIR", str(tmp_path))
from utils.paths import REPO_ROOT
assert analytics._cs_dir("etfs") == REPO_ROOT / "case_studies"
def test_registry_path_is_three_levels_deep(tmp_path, monkeypatch) -> None:
monkeypatch.setenv("ML4T_OUTPUT_DIR", str(tmp_path))
(tmp_path / "etfs" / "run_log").mkdir(parents=True)
(tmp_path / "etfs" / "run_log" / "registry.db").touch()
p = analytics._registry_path("etfs")
assert p == tmp_path / "etfs" / "run_log" / "registry.db"
# -----------------------------------------------------------------------------
# _query behavior on empty / missing databases
# -----------------------------------------------------------------------------
def test_query_returns_empty_df_when_path_missing(tmp_path) -> None:
missing = tmp_path / "nope.db"
out = analytics._query(missing, "SELECT 1")
assert isinstance(out, pl.DataFrame)
assert out.is_empty()
def test_query_returns_empty_df_when_no_rows(tmp_path) -> None:
db = tmp_path / "empty.db"
conn = sqlite3.connect(str(db))
conn.execute("CREATE TABLE t (x INTEGER)")
conn.commit()
conn.close()
out = analytics._query(db, "SELECT * FROM t")
assert out.is_empty()
def test_query_returns_populated_df(tmp_path) -> None:
db = tmp_path / "rows.db"
conn = sqlite3.connect(str(db))
conn.execute("CREATE TABLE t (x INTEGER, y TEXT)")
conn.executemany("INSERT INTO t VALUES (?, ?)", [(1, "a"), (2, "b")])
conn.commit()
conn.close()
out = analytics._query(db, "SELECT * FROM t ORDER BY x")
assert out.to_dicts() == [{"x": 1, "y": "a"}, {"x": 2, "y": "b"}]
# -----------------------------------------------------------------------------
# Seeded-registry fixture: builds the schema + minimal rows
# -----------------------------------------------------------------------------
def _create_registry_schema(conn: sqlite3.Connection) -> None:
"""Create the subset of the registry schema the analytics queries touch."""
conn.executescript(
"""
CREATE TABLE training_runs (
training_hash TEXT PRIMARY KEY,
family TEXT NOT NULL,
label TEXT NOT NULL,
config_name TEXT,
spec_json TEXT,
created_at TEXT NOT NULL
);
CREATE TABLE prediction_sets (
prediction_hash TEXT PRIMARY KEY,
training_hash TEXT NOT NULL,
checkpoint_value INTEGER,
checkpoint_kind TEXT,
split TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE prediction_metrics (
prediction_hash TEXT PRIMARY KEY,
computed_at TEXT NOT NULL,
ic_mean REAL,
ic_std REAL,
ic_t REAL,
n_folds REAL,
pct_positive REAL,
task_type TEXT,
accuracy REAL,
balanced_accuracy REAL,
auc_roc REAL,
auc_pr REAL,
log_loss REAL,
brier_score REAL
);
CREATE TABLE backtest_runs (
backtest_hash TEXT PRIMARY KEY,
prediction_hash TEXT NOT NULL,
spec_json TEXT,
stage TEXT,
created_at TEXT NOT NULL
);
CREATE TABLE backtest_metrics (
backtest_hash TEXT PRIMARY KEY,
computed_at TEXT NOT NULL,
sharpe REAL,
sortino REAL,
total_return REAL,
max_drawdown REAL
);
CREATE TABLE fold_metrics (
prediction_hash TEXT NOT NULL,
fold_id INTEGER NOT NULL,
computed_at TEXT NOT NULL,
ic REAL,
PRIMARY KEY (prediction_hash, fold_id)
);
"""
)
def _seed_registry(db_path: Path) -> None:
"""Populate a registry with 4 training runs + predictions + 2 backtests.
Layout (etfs):
linear_a / fwd_ret_21d / validation / ic=0.05 / regression
linear_b / fwd_ret_21d / validation / ic=0.08 / regression <- best linear
gbm_a / fwd_ret_21d / validation / ic=0.10 / regression <- best gbm (primary)
gbm_a / fwd_ret_21d / holdout / ic=0.03 / regression
linear_c / fwd_dir_5d / validation / ic=0.04 / classification (task_type='classification')
Plus 1 ch16 (signal) backtest and 1 ch17 (allocation) backtest on the
same gbm prediction_hash.
"""
db_path.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(db_path))
_create_registry_schema(conn)
# training_runs
conn.executemany(
"INSERT INTO training_runs VALUES (?, ?, ?, ?, ?, ?)",
[
("th_lin_a", "linear", "fwd_ret_21d", "ridge_a100", None, "2024-01-01T00:00:00"),
("th_lin_b", "linear", "fwd_ret_21d", "ridge_b100", None, "2024-01-01T00:00:00"),
("th_gbm_a", "gbm", "fwd_ret_21d", "default", None, "2024-01-01T00:00:00"),
("th_lin_c", "linear", "fwd_dir_5d", "logistic", None, "2024-01-01T00:00:00"),
],
)
# prediction_sets
conn.executemany(
"INSERT INTO prediction_sets VALUES (?, ?, ?, ?, ?, ?)",
[
("ph_lin_a_val", "th_lin_a", 0, "final", "validation", "2024-01-02T00:00:00"),
("ph_lin_b_val", "th_lin_b", 0, "final", "validation", "2024-01-02T00:00:00"),
("ph_gbm_a_val", "th_gbm_a", 100, "final", "validation", "2024-01-02T00:00:00"),
("ph_gbm_a_hol", "th_gbm_a", 100, "final", "holdout", "2024-01-02T00:00:00"),
("ph_lin_c_val", "th_lin_c", 0, "final", "validation", "2024-01-02T00:00:00"),
],
)
# prediction_metrics — regression and classification
conn.executemany(
"INSERT INTO prediction_metrics (prediction_hash, computed_at, ic_mean, task_type, "
"auc_roc, accuracy) VALUES (?, ?, ?, ?, ?, ?)",
[
("ph_lin_a_val", "2024-01-03", 0.05, "regression", None, None),
("ph_lin_b_val", "2024-01-03", 0.08, "regression", None, None),
("ph_gbm_a_val", "2024-01-03", 0.10, "regression", None, None),
("ph_gbm_a_hol", "2024-01-03", 0.03, "regression", None, None),
("ph_lin_c_val", "2024-01-03", 0.04, "classification", 0.62, 0.55),
],
)
# backtest_runs — one signal (ch16) + one allocation (ch17) on the gbm prediction
spec_v2 = {
"version": 2,
"strategy": {"allocation": {"method": "inverse_vol"}},
"backtest_config": {
"commission": {"rate": 0.0005}, # 5 bps
"slippage": {"rate": 0.0003}, # 3 bps
},
}
conn.executemany(
"INSERT INTO backtest_runs VALUES (?, ?, ?, ?, ?)",
[
("bh_sig", "ph_gbm_a_val", json.dumps(spec_v2), "signal", "2024-01-04"),
("bh_alloc", "ph_gbm_a_val", json.dumps(spec_v2), "allocation", "2024-01-04"),
],
)
conn.executemany(
"INSERT INTO backtest_metrics (backtest_hash, computed_at, sharpe, sortino, "
"total_return, max_drawdown) VALUES (?, ?, ?, ?, ?, ?)",
[
("bh_sig", "2024-01-05", 1.2, 1.8, 0.35, -0.10),
("bh_alloc", "2024-01-05", 1.5, 2.2, 0.45, -0.08),
],
)
conn.commit()
conn.close()
@pytest.fixture
def seeded_registries(tmp_path, monkeypatch) -> Path:
"""Build registries for etfs + crypto_perps_funding under a temp output dir.
The second case study (crypto) is intentionally empty (only schema) so
multi-case-study queries have a no-op partition to merge against.
"""
monkeypatch.setenv("ML4T_OUTPUT_DIR", str(tmp_path))
_seed_registry(tmp_path / "etfs" / "run_log" / "registry.db")
# crypto: schema but no rows
crypto_db = tmp_path / "crypto_perps_funding" / "run_log" / "registry.db"
crypto_db.parent.mkdir(parents=True)
conn = sqlite3.connect(str(crypto_db))
_create_registry_schema(conn)
conn.commit()
conn.close()
return tmp_path
# -----------------------------------------------------------------------------
# load_model_ic
# -----------------------------------------------------------------------------
def test_load_model_ic_returns_all_families_by_default(seeded_registries) -> None:
df = analytics.load_model_ic(case_studies=["etfs", "crypto_perps_funding"], split="validation")
# etfs: 3 validation rows (lin_a, lin_b, gbm_a) with regression task_type;
# lin_c is also validation but the query doesn't filter task_type here.
assert df.height == 4
assert set(df["case_study"].unique().to_list()) == {"etfs"}
assert set(df["family"].unique().to_list()) == {"linear", "gbm"}
def test_load_model_ic_filters_by_family(seeded_registries) -> None:
df = analytics.load_model_ic(families="gbm", case_studies=["etfs"], split="validation")
assert df["family"].unique().to_list() == ["gbm"]
assert df.height == 1
def test_load_model_ic_filters_by_split_holdout(seeded_registries) -> None:
df = analytics.load_model_ic(case_studies=["etfs"], split="holdout")
# only gbm_a has a holdout prediction
assert df.height == 1
assert df["split"].to_list() == ["holdout"]
assert df["ic_mean"].to_list() == [0.03]
def test_load_model_ic_returns_empty_when_no_case_study_has_data(seeded_registries) -> None:
df = analytics.load_model_ic(case_studies=["crypto_perps_funding"], split="validation")
assert df.is_empty()
def test_load_model_ic_has_case_study_label_column(seeded_registries) -> None:
df = analytics.load_model_ic(case_studies=["etfs"], split="validation")
assert "case_study" in df.columns
assert df["case_study"].unique().to_list() == ["etfs"]
# -----------------------------------------------------------------------------
# load_classification_metrics
# -----------------------------------------------------------------------------
def test_load_classification_metrics_filters_task_type_eq_1(seeded_registries) -> None:
"""Only the linear_c row (task_type='classification') should come back."""
df = analytics.load_classification_metrics(case_studies=["etfs"], split="validation")
assert df.height == 1
assert df["family"].to_list() == ["linear"]
assert df["auc_roc"].to_list() == [0.62]
assert df["task_type"].to_list() == ["classification"]
def test_load_classification_metrics_excludes_regression_rows(seeded_registries) -> None:
"""Regression rows (task_type='regression') must not leak into the classification view."""
df = analytics.load_classification_metrics(case_studies=["etfs"], split="validation")
# Spec: no rows with null AUC should appear
assert df.filter(pl.col("auc_roc").is_null()).is_empty()
# -----------------------------------------------------------------------------
# load_best_ic_per_family
# -----------------------------------------------------------------------------
def test_load_best_ic_per_family_picks_max_per_pair(seeded_registries) -> None:
"""Primary label for etfs is fwd_ret_21d. Among linear runs on that label,
lin_b (IC=0.08) beats lin_a (IC=0.05). gbm has only one run (IC=0.10).
"""
best = analytics.load_best_ic_per_family(case_studies=["etfs"], split="validation")
rows = {(r["case_study"], r["family"]): r for r in best.to_dicts()}
assert rows[("etfs", "linear")]["ic_mean"] == 0.08
assert rows[("etfs", "linear")]["config_name"] == "ridge_b100"
assert rows[("etfs", "gbm")]["ic_mean"] == 0.10
def test_load_best_ic_per_family_primary_label_excludes_other_labels(
seeded_registries,
) -> None:
"""With use_primary_label=True (default), the linear_c run on fwd_dir_5d must
not appear — only rows with label == primary are kept.
"""
best = analytics.load_best_ic_per_family(case_studies=["etfs"], split="validation")
assert all(row["label"] == "fwd_ret_21d" for row in best.to_dicts())
def test_load_best_ic_per_family_use_primary_label_false_includes_all(
seeded_registries,
) -> None:
"""With use_primary_label=False, the fwd_dir_5d row competes for best linear."""
best = analytics.load_best_ic_per_family(
case_studies=["etfs"], split="validation", use_primary_label=False
)
# linear: best of {lin_a 0.05, lin_b 0.08, lin_c 0.04} is lin_b
linear_row = next(r for r in best.to_dicts() if r["family"] == "linear")
assert linear_row["ic_mean"] == 0.08
def test_load_best_ic_per_family_adds_display_name(seeded_registries) -> None:
best = analytics.load_best_ic_per_family(case_studies=["etfs"], split="validation")
assert "display_name" in best.columns
assert best["display_name"].unique().to_list() == ["ETFs"]
# -----------------------------------------------------------------------------
# load_chapter_backtests
# -----------------------------------------------------------------------------
def test_load_chapter_backtests_ch16_maps_to_signal_stage(seeded_registries) -> None:
"""chapter='ch16' → stage='signal' → returns the bh_sig backtest row."""
df = analytics.load_chapter_backtests("ch16", case_studies=["etfs"])
assert df.height == 1
assert df["backtest_hash"].to_list() == ["bh_sig"]
def test_load_chapter_backtests_explicit_stage_overrides_chapter(seeded_registries) -> None:
df = analytics.load_chapter_backtests("ch16", stage="allocation", case_studies=["etfs"])
assert df["backtest_hash"].to_list() == ["bh_alloc"]
def test_load_chapter_backtests_joins_sharpe_and_training_columns(seeded_registries) -> None:
df = analytics.load_chapter_backtests("ch17", case_studies=["etfs"])
assert df.height == 1
row = df.to_dicts()[0]
assert row["sharpe"] == 1.5
assert row["family"] == "gbm"
assert row["config_name"] == "default"
def test_load_chapter_backtests_metrics_filter_selects_columns(seeded_registries) -> None:
df = analytics.load_chapter_backtests("ch17", case_studies=["etfs"], metrics=["sharpe"])
# Only meta columns + sharpe; sortino/total_return/max_drawdown excluded
assert "sharpe" in df.columns
assert "sortino" not in df.columns
assert "total_return" not in df.columns
def test_load_chapter_backtests_returns_empty_for_unused_stage(seeded_registries) -> None:
df = analytics.load_chapter_backtests("ch18", case_studies=["etfs"])
assert df.is_empty()
# -----------------------------------------------------------------------------
# Spec helpers
# -----------------------------------------------------------------------------
def test_parse_backtest_spec_round_trips_json() -> None:
spec = {"version": 2, "strategy": {"foo": "bar"}, "backtest_config": {}}
assert analytics.parse_backtest_spec(json.dumps(spec)) == spec
def test_extract_cost_bps_sums_commission_and_slippage_from_v2_spec() -> None:
spec_json = json.dumps(
{
"version": 2,
"strategy": {},
"backtest_config": {
"commission": {"rate": 0.0005}, # 5 bps
"slippage": {"rate": 0.0003}, # 3 bps
},
}
)
assert analytics.extract_cost_bps(spec_json) == pytest.approx(8.0)
def test_extract_cost_bps_handles_v1_spec() -> None:
"""v1 specs store costs in a flat ``costs`` dict; cost_view falls back to it."""
spec_json = json.dumps({"costs": {"commission_bps": 2.0, "slippage_bps": 1.5}})
assert analytics.extract_cost_bps(spec_json) == pytest.approx(3.5)
def test_extract_allocator_reads_strategy_allocation_method() -> None:
spec_json = json.dumps(
{
"version": 2,
"strategy": {"allocation": {"method": "risk_parity"}},
"backtest_config": {},
}
)
assert analytics.extract_allocator(spec_json) == "risk_parity"
def test_extract_allocator_defaults_to_unknown_when_missing() -> None:
spec_json = json.dumps({"version": 2, "strategy": {}, "backtest_config": {}})
assert analytics.extract_allocator(spec_json) == "unknown"
在新工单中引用 查看 Git Blame 复制永久链接