wehub-resource-sync
614 lines
21 KiB
Python
614 lines
21 KiB
Python
"""Cross-dataset analytics for case study insights notebooks.
|
|
|
|
Replaces per-notebook JSON-loading patterns with direct registry queries.
|
|
Each function returns a Polars DataFrame ready for analysis and visualization.
|
|
|
|
Usage::
|
|
|
|
from case_studies.utils.analytics import (
|
|
CASE_STUDY_IDS, PRIMARY_LABELS, SHORT_NAMES, DATASET_META,
|
|
load_model_ic, load_best_ic_per_family, load_chapter_backtests,
|
|
)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
import polars as pl
|
|
|
|
from case_studies.utils.backtest_presets import cost_view, strategy_view
|
|
from case_studies.utils.notebook_contracts import degenerate_prediction_sql
|
|
from utils.paths import REPO_ROOT
|
|
|
|
CASE_STUDY_META = {
|
|
"etfs": {"display_name": "ETFs", "chapter_track": "Ch6 to Ch21"},
|
|
"crypto_perps_funding": {
|
|
"display_name": "Crypto Perps Funding",
|
|
"chapter_track": "Ch6 to Ch12",
|
|
},
|
|
"nasdaq100_microstructure": {
|
|
"display_name": "NASDAQ-100 Microstructure",
|
|
"chapter_track": "Ch6 to Ch12",
|
|
},
|
|
"sp500_equity_option_analytics": {
|
|
"display_name": "S&P 500 Equity+Options",
|
|
"chapter_track": "Ch6 to Ch21",
|
|
},
|
|
"us_firm_characteristics": {
|
|
"display_name": "US Firm Characteristics",
|
|
"chapter_track": "Ch6 to Ch14",
|
|
},
|
|
"fx_pairs": {"display_name": "FX Pairs", "chapter_track": "Ch6 to Ch17"},
|
|
"cme_futures": {"display_name": "CME Futures", "chapter_track": "Ch6 to Ch17"},
|
|
"sp500_options": {
|
|
"display_name": "S&P 500 Options",
|
|
"chapter_track": "Ch6 to Ch21",
|
|
},
|
|
"us_equities_panel": {
|
|
"display_name": "US Equities Panel",
|
|
"chapter_track": "Ch6 to Ch14",
|
|
},
|
|
}
|
|
|
|
DISPLAY_NAMES = {k: v["display_name"] for k, v in CASE_STUDY_META.items()}
|
|
CHAPTER_TRACKS = {k: v["chapter_track"] for k, v in CASE_STUDY_META.items()}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Canonical metadata (single source — replaces duplicated dicts in notebooks)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
CASE_STUDY_IDS = list(CASE_STUDY_META.keys())
|
|
|
|
PRIMARY_LABELS = {
|
|
"etfs": "fwd_ret_21d",
|
|
"crypto_perps_funding": "fwd_ret_8h",
|
|
"nasdaq100_microstructure": "fwd_ret_15m",
|
|
"sp500_equity_option_analytics": "fwd_ret_5d",
|
|
"us_firm_characteristics": "fwd_ret_1m",
|
|
"fx_pairs": "fwd_ret_1d",
|
|
"cme_futures": "fwd_ret_5d",
|
|
"sp500_options": "ret_to_expiry",
|
|
"us_equities_panel": "fwd_ret_1d",
|
|
}
|
|
|
|
SHORT_NAMES = {
|
|
"etfs": "ETFs",
|
|
"crypto_perps_funding": "Crypto",
|
|
"nasdaq100_microstructure": "NQ100",
|
|
"sp500_equity_option_analytics": "SP500 Eq+Opt",
|
|
"us_firm_characteristics": "US Firms",
|
|
"fx_pairs": "FX",
|
|
"cme_futures": "CME Futures",
|
|
"sp500_options": "SP500 Options",
|
|
"us_equities_panel": "US Equities",
|
|
}
|
|
|
|
DATASET_META = {
|
|
"etfs": {"frequency": "Daily", "entities": 99, "horizon": "21d"},
|
|
"crypto_perps_funding": {"frequency": "8-hourly", "entities": 21, "horizon": "8h"},
|
|
"nasdaq100_microstructure": {"frequency": "15-min", "entities": 114, "horizon": "15m"},
|
|
"sp500_equity_option_analytics": {"frequency": "Daily", "entities": 638, "horizon": "5d"},
|
|
"us_firm_characteristics": {"frequency": "Monthly", "entities": 2483, "horizon": "1m"},
|
|
"fx_pairs": {"frequency": "Daily", "entities": 20, "horizon": "1d"},
|
|
"cme_futures": {"frequency": "Daily", "entities": 30, "horizon": "5d"},
|
|
"sp500_options": {"frequency": "Daily", "entities": 612, "horizon": "dh-10d"},
|
|
"us_equities_panel": {"frequency": "Daily", "entities": 3199, "horizon": "1d"},
|
|
}
|
|
|
|
CADENCE_MAP = {
|
|
"etfs": "monthly",
|
|
"crypto_perps_funding": "8-hourly",
|
|
"nasdaq100_microstructure": "15-min",
|
|
"cme_futures": "monthly",
|
|
"fx_pairs": "daily",
|
|
"sp500_equity_option_analytics": "monthly",
|
|
"sp500_options": "monthly",
|
|
"us_equities_panel": "daily",
|
|
"us_firm_characteristics": "monthly",
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Internal helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _cs_dir(case_study: str | None = None) -> Path:
|
|
"""Resolve case study root, respecting ML4T_OUTPUT_DIR for test isolation."""
|
|
import os
|
|
|
|
output_dir = os.environ.get("ML4T_OUTPUT_DIR")
|
|
if output_dir:
|
|
base = Path(output_dir)
|
|
if case_study and (base / case_study / "run_log" / "registry.db").exists():
|
|
return base
|
|
return REPO_ROOT / "case_studies"
|
|
|
|
|
|
def _registry_path(case_study: str) -> Path:
|
|
return _cs_dir(case_study) / case_study / "run_log" / "registry.db"
|
|
|
|
|
|
def _query(db_path: Path, sql: str, params: tuple = ()) -> pl.DataFrame:
|
|
"""Execute SQL on a registry and return a Polars DataFrame."""
|
|
if not db_path.exists():
|
|
return pl.DataFrame()
|
|
conn = sqlite3.connect(str(db_path))
|
|
conn.row_factory = sqlite3.Row
|
|
try:
|
|
rows = conn.execute(sql, params).fetchall()
|
|
if not rows:
|
|
return pl.DataFrame()
|
|
return pl.DataFrame([dict(r) for r in rows], infer_schema_length=None)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Model IC queries
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def load_model_ic(
|
|
families: list[str] | str | None = None,
|
|
*,
|
|
split: str = "validation",
|
|
case_studies: list[str] | None = None,
|
|
) -> pl.DataFrame:
|
|
"""Load IC metrics across case studies for specified model families.
|
|
|
|
Returns a DataFrame with columns:
|
|
case_study, family, config_name, label, split,
|
|
checkpoint_value, ic_mean, ic_std
|
|
|
|
Parameters
|
|
----------
|
|
families : str or list of str, optional
|
|
Filter to specific families (e.g. "linear", ["gbm", "deep_learning"]).
|
|
None returns all families.
|
|
split : str
|
|
Prediction split to filter by ("validation" or "holdout").
|
|
case_studies : list of str, optional
|
|
Case studies to query. None = all.
|
|
"""
|
|
if isinstance(families, str):
|
|
families = [families]
|
|
cs_list = case_studies or CASE_STUDY_IDS
|
|
|
|
frames = []
|
|
for cs_id in cs_list:
|
|
db_path = _registry_path(cs_id)
|
|
if not db_path.exists():
|
|
continue
|
|
|
|
family_clause = ""
|
|
params: list = []
|
|
if families:
|
|
placeholders = ",".join("?" * len(families))
|
|
family_clause = f"AND t.family IN ({placeholders})"
|
|
params.extend(families)
|
|
|
|
params.append(split)
|
|
|
|
sql = f"""
|
|
SELECT
|
|
t.family,
|
|
t.config_name,
|
|
t.label,
|
|
p.split,
|
|
p.checkpoint_value,
|
|
p.prediction_hash,
|
|
pm.ic_mean,
|
|
pm.ic_std
|
|
FROM training_runs t
|
|
JOIN prediction_sets p ON t.training_hash = p.training_hash
|
|
JOIN prediction_metrics pm ON p.prediction_hash = pm.prediction_hash
|
|
WHERE 1=1 {family_clause}
|
|
AND p.split = ?
|
|
{degenerate_prediction_sql("p.prediction_hash")}
|
|
ORDER BY pm.ic_mean DESC NULLS LAST
|
|
"""
|
|
df = _query(db_path, sql, tuple(params))
|
|
if len(df) > 0:
|
|
# Cast columns to expected types (handles Null-type columns from empty aggregates)
|
|
col_types = {"checkpoint_value": pl.Int64, "ic_mean": pl.Float64, "ic_std": pl.Float64}
|
|
casts = [
|
|
pl.col(c).cast(t, strict=False) for c, t in col_types.items() if c in df.columns
|
|
]
|
|
if casts:
|
|
df = df.with_columns(casts)
|
|
frames.append(df.with_columns(pl.lit(cs_id).alias("case_study")))
|
|
|
|
if not frames:
|
|
return pl.DataFrame()
|
|
return pl.concat(frames, how="diagonal")
|
|
|
|
|
|
def resolve_best_prediction(
|
|
case_study: str,
|
|
label: str,
|
|
*,
|
|
family: str = "gbm",
|
|
split: str = "validation",
|
|
) -> dict:
|
|
"""Return rank-1 prediction-set metadata for (case_study, label, family).
|
|
|
|
Resolves the best-IC prediction set from `prediction_metrics` (sorted by
|
|
`ic_mean` descending). Useful for downstream notebooks (Ch16/17) that need
|
|
to consume registered upstream predictions without baking in a hash.
|
|
|
|
Returns
|
|
-------
|
|
dict
|
|
Keys: prediction_hash, config_name, ic_mean, ic_std, family, label, split.
|
|
|
|
Raises
|
|
------
|
|
RuntimeError
|
|
If no prediction set matches (case_study, label, family, split).
|
|
"""
|
|
df = (
|
|
load_model_ic([family], split=split, case_studies=[case_study])
|
|
.filter(pl.col("label") == label)
|
|
.filter(pl.col("ic_mean").is_not_null())
|
|
.sort("ic_mean", descending=True)
|
|
)
|
|
if df.is_empty():
|
|
raise RuntimeError(
|
|
f"No {family} predictions with non-null ic_mean for "
|
|
f"{case_study}/{label}/{split} in registry.db. "
|
|
f"Run the {family} training notebook for {case_study} before this notebook."
|
|
)
|
|
return df.row(0, named=True)
|
|
|
|
|
|
def load_classification_metrics(
|
|
families: list[str] | str | None = None,
|
|
*,
|
|
split: str = "validation",
|
|
case_studies: list[str] | None = None,
|
|
) -> pl.DataFrame:
|
|
"""Load classification metrics (AUC-ROC, accuracy, etc.) from registries.
|
|
|
|
Returns a DataFrame with columns:
|
|
case_study, family, config_name, label, split,
|
|
ic_mean, auc_roc, accuracy, balanced_accuracy, log_loss, brier_score, auc_pr
|
|
"""
|
|
if isinstance(families, str):
|
|
families = [families]
|
|
cs_list = case_studies or CASE_STUDY_IDS
|
|
|
|
frames = []
|
|
for cs_id in cs_list:
|
|
db_path = _registry_path(cs_id)
|
|
if not db_path.exists():
|
|
continue
|
|
|
|
family_clause = ""
|
|
params: list = []
|
|
if families:
|
|
placeholders = ",".join("?" * len(families))
|
|
family_clause = f"AND t.family IN ({placeholders})"
|
|
params.extend(families)
|
|
|
|
params.append(split)
|
|
|
|
sql = f"""
|
|
SELECT
|
|
t.family,
|
|
t.config_name,
|
|
t.label,
|
|
p.split,
|
|
pm.ic_mean,
|
|
pm.auc_roc,
|
|
pm.accuracy,
|
|
pm.balanced_accuracy,
|
|
pm.log_loss,
|
|
pm.brier_score,
|
|
pm.auc_pr,
|
|
pm.task_type
|
|
FROM training_runs t
|
|
JOIN prediction_sets p ON t.training_hash = p.training_hash
|
|
JOIN prediction_metrics pm ON p.prediction_hash = pm.prediction_hash
|
|
WHERE 1=1 {family_clause}
|
|
AND p.split = ?
|
|
AND pm.task_type = 'classification'
|
|
ORDER BY pm.auc_roc DESC NULLS LAST
|
|
"""
|
|
df = _query(db_path, sql, tuple(params))
|
|
if len(df) > 0:
|
|
frames.append(df.with_columns(pl.lit(cs_id).alias("case_study")))
|
|
|
|
if not frames:
|
|
return pl.DataFrame()
|
|
return pl.concat(frames, how="diagonal")
|
|
|
|
|
|
def load_best_ic_per_family(
|
|
families: list[str] | None = None,
|
|
*,
|
|
split: str = "validation",
|
|
case_studies: list[str] | None = None,
|
|
use_primary_label: bool = True,
|
|
) -> pl.DataFrame:
|
|
"""Best IC per family per case study (one row per family-case_study pair).
|
|
|
|
Returns: case_study, display_name, family, config_name, label, ic_mean
|
|
"""
|
|
all_ic = load_model_ic(families, split=split, case_studies=case_studies)
|
|
if all_ic.is_empty():
|
|
return pl.DataFrame()
|
|
|
|
if use_primary_label:
|
|
label_rows = [
|
|
{"case_study": cs, "primary_label": lbl} for cs, lbl in PRIMARY_LABELS.items()
|
|
]
|
|
label_df = pl.DataFrame(label_rows)
|
|
all_ic = all_ic.join(label_df, on="case_study").filter(
|
|
pl.col("label") == pl.col("primary_label")
|
|
)
|
|
|
|
best = (
|
|
all_ic.sort("ic_mean", descending=True, nulls_last=True)
|
|
.group_by(["case_study", "family"])
|
|
.first()
|
|
.select("case_study", "family", "config_name", "label", "ic_mean")
|
|
)
|
|
|
|
# Add display names
|
|
name_rows = [{"case_study": k, "display_name": v} for k, v in SHORT_NAMES.items()]
|
|
name_df = pl.DataFrame(name_rows)
|
|
best = best.join(name_df, on="case_study", how="left")
|
|
|
|
return best.sort(["case_study", "family"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Backtest queries
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
_CHAPTER_TO_STAGE = {
|
|
"ch16": "signal",
|
|
"ch17": "allocation",
|
|
"ch18": "cost_sensitivity",
|
|
"ch19": "risk_overlay",
|
|
}
|
|
|
|
|
|
def load_chapter_backtests(
|
|
chapter: str,
|
|
*,
|
|
stage: str | None = None,
|
|
case_studies: list[str] | None = None,
|
|
metrics: list[str] | None = None,
|
|
) -> pl.DataFrame:
|
|
"""Load backtest results for a pipeline stage across case studies.
|
|
|
|
Returns a DataFrame with columns:
|
|
case_study, display_name, backtest_hash, prediction_hash,
|
|
family, config_name, label, spec_json,
|
|
plus one column per metric (sharpe, sortino, etc.)
|
|
|
|
Parameters
|
|
----------
|
|
chapter : str
|
|
Legacy chapter tag (e.g. "ch16"). Mapped to stage automatically.
|
|
stage : str, optional
|
|
Pipeline stage to filter by. Overrides chapter mapping.
|
|
case_studies : list of str, optional
|
|
Case studies to query. None = all.
|
|
metrics : list of str, optional
|
|
Metrics to pivot. None = all available.
|
|
"""
|
|
cs_list = case_studies or CASE_STUDY_IDS
|
|
resolved_stage = stage or _CHAPTER_TO_STAGE.get(chapter, chapter)
|
|
|
|
frames = []
|
|
for cs_id in cs_list:
|
|
db_path = _registry_path(cs_id)
|
|
if not db_path.exists():
|
|
continue
|
|
|
|
# Prefer stage column; fall back to spec_json LIKE for un-migrated DBs
|
|
sql = """
|
|
SELECT
|
|
b.backtest_hash,
|
|
b.prediction_hash,
|
|
b.spec_json,
|
|
t.family,
|
|
t.config_name,
|
|
t.label,
|
|
bm.*
|
|
FROM backtest_runs b
|
|
JOIN backtest_metrics bm ON b.backtest_hash = bm.backtest_hash
|
|
JOIN prediction_sets p ON b.prediction_hash = p.prediction_hash
|
|
JOIN training_runs t ON p.training_hash = t.training_hash
|
|
WHERE b.stage = ?
|
|
"""
|
|
df = _query(db_path, sql, (resolved_stage,))
|
|
if len(df) > 0:
|
|
frames.append(df.with_columns(pl.lit(cs_id).alias("case_study")))
|
|
|
|
if not frames:
|
|
return pl.DataFrame()
|
|
|
|
result = pl.concat(frames, how="diagonal_relaxed")
|
|
|
|
# Filter to requested metrics (select only those columns)
|
|
if metrics:
|
|
# Keep meta columns + requested metric columns
|
|
meta_cols = [
|
|
"case_study",
|
|
"backtest_hash",
|
|
"prediction_hash",
|
|
"spec_json",
|
|
"family",
|
|
"config_name",
|
|
"label",
|
|
]
|
|
keep = [c for c in meta_cols if c in result.columns]
|
|
keep += [c for c in metrics if c in result.columns]
|
|
result = result.select(keep)
|
|
|
|
# Drop internal columns from bm.* join
|
|
drop_cols = [c for c in ["computed_at"] if c in result.columns]
|
|
if drop_cols:
|
|
result = result.drop(drop_cols)
|
|
|
|
# Add display names
|
|
name_rows = [{"case_study": k, "display_name": v} for k, v in SHORT_NAMES.items()]
|
|
name_df = pl.DataFrame(name_rows)
|
|
result = result.join(name_df, on="case_study", how="left")
|
|
|
|
return result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Spec parsing helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def parse_backtest_spec(spec_json: str) -> dict:
|
|
"""Parse a backtest spec_json string into a dict."""
|
|
return json.loads(spec_json)
|
|
|
|
|
|
def extract_cost_bps(spec_json: str) -> float:
|
|
"""Extract total cost (commission + slippage) in bps from a backtest spec."""
|
|
spec = parse_backtest_spec(spec_json)
|
|
costs = cost_view(spec)
|
|
return float(costs.get("commission_bps", 0.0)) + float(costs.get("slippage_bps", 0.0))
|
|
|
|
|
|
def extract_allocator(spec_json: str) -> str:
|
|
"""Extract allocation method from a backtest spec."""
|
|
spec = parse_backtest_spec(spec_json)
|
|
allocation = strategy_view(spec).get("allocation") or {}
|
|
return allocation.get("method", "unknown")
|
|
|
|
|
|
# nasdaq100_microstructure deploys the cost-feasible *ensemble* carrier
|
|
# (developed in 20.4). resolve_canonical_rank1_lineage returns the
|
|
# non-deployed full-universe val-max gbm instead, so the ensemble's training
|
|
# hash and validation backtest are pinned here. See agents
|
|
# UNCERTAINTY_ARCHITECTURE / Ch20 audit.
|
|
_NASDAQ_ENSEMBLE_TRAINING_HASH = "a9f04b886b9a"
|
|
_NASDAQ_ENSEMBLE_VAL_HASH = "4e939dee0a5f"
|
|
|
|
|
|
def _strategy_signature(spec_json: str) -> str:
|
|
"""Identity of a backtest's signal + allocation config, ignoring cost.
|
|
|
|
The cost sweep holds the strategy fixed and varies only commission +
|
|
slippage, so every cost-grid run for one carrier shares this signature.
|
|
A single training hash can host more than one carrier (e.g. a signal-only
|
|
eq-weight series alongside an allocator series); matching the signature
|
|
selects the deployed one rather than a sibling.
|
|
"""
|
|
sv = strategy_view(parse_backtest_spec(spec_json))
|
|
return json.dumps([sv.get("signal"), sv.get("allocation")], sort_keys=True)
|
|
|
|
|
|
def load_carrier_cost_curves(case_studies: list[str] | None = None) -> pl.DataFrame:
|
|
"""Cost-sensitivity curves for each case study's *deployed carrier*.
|
|
|
|
The carrier is the highest-validation-Sharpe configuration across the
|
|
signal, allocation, and risk-overlay stages, resolved via
|
|
``resolve_canonical_rank1_lineage``. The cost sweep (Ch18) holds that
|
|
carrier's signal and allocation fixed while varying commission +
|
|
slippage, so the breakeven implied here is the carrier's own cost
|
|
survival — not that of whichever allocator happened to be best at zero
|
|
cost (which need not be the deployed strategy). Rows are matched to the
|
|
carrier's exact strategy signature, so a sibling series sharing the
|
|
training hash (e.g. a signal-only eq-weight run) is excluded.
|
|
|
|
Returns tidy rows ``[case_study, display_name, cadence, label,
|
|
allocator, cost_bps, sharpe, total_return, max_drawdown]`` on the
|
|
validation split, sorted by ``cost_bps`` within each case study.
|
|
"""
|
|
# Lazy import avoids a module-load cycle (strategy_analysis is heavier).
|
|
from case_studies.utils.strategy_analysis import resolve_canonical_rank1_lineage
|
|
|
|
cs_list = case_studies or CASE_STUDY_IDS
|
|
cost_sql = """
|
|
SELECT b.spec_json, t.label, bm.sharpe, bm.total_return, bm.max_drawdown
|
|
FROM backtest_runs b
|
|
JOIN backtest_metrics bm ON b.backtest_hash = bm.backtest_hash
|
|
JOIN prediction_sets p ON b.prediction_hash = p.prediction_hash
|
|
JOIN training_runs t ON p.training_hash = t.training_hash
|
|
WHERE b.stage = 'cost_sensitivity' AND p.split = 'validation'
|
|
AND p.training_hash = ?
|
|
"""
|
|
|
|
frames = []
|
|
for cs_id in cs_list:
|
|
db_path = _registry_path(cs_id)
|
|
if not db_path.exists():
|
|
continue
|
|
|
|
# Resolve the deployed carrier's training hash and its strategy spec.
|
|
if cs_id == "nasdaq100_microstructure":
|
|
training_hash = _NASDAQ_ENSEMBLE_TRAINING_HASH
|
|
spec_df = _query(
|
|
db_path,
|
|
"SELECT spec_json FROM backtest_runs WHERE backtest_hash LIKE ?",
|
|
(_NASDAQ_ENSEMBLE_VAL_HASH + "%",),
|
|
)
|
|
else:
|
|
try:
|
|
lin = resolve_canonical_rank1_lineage(cs_id)
|
|
except Exception:
|
|
continue
|
|
training_hash = lin.get("training_hash")
|
|
spec_df = _query(
|
|
db_path,
|
|
"SELECT spec_json FROM backtest_runs WHERE backtest_hash = ?",
|
|
(lin.get("val_backtest_hash"),),
|
|
)
|
|
if not training_hash or spec_df.is_empty():
|
|
continue
|
|
carrier_sig = _strategy_signature(spec_df["spec_json"][0])
|
|
|
|
df = _query(db_path, cost_sql, (training_hash,))
|
|
if df.is_empty():
|
|
continue
|
|
df = df.with_columns(
|
|
cost_bps=pl.col("spec_json").map_elements(extract_cost_bps, return_dtype=pl.Float64),
|
|
allocator=pl.col("spec_json").map_elements(extract_allocator, return_dtype=pl.Utf8),
|
|
signature=pl.col("spec_json").map_elements(_strategy_signature, return_dtype=pl.Utf8),
|
|
case_study=pl.lit(cs_id),
|
|
).filter(pl.col("signature") == carrier_sig)
|
|
if df.is_empty():
|
|
continue
|
|
frames.append(
|
|
df.select(
|
|
"case_study",
|
|
"label",
|
|
"allocator",
|
|
"cost_bps",
|
|
"sharpe",
|
|
"total_return",
|
|
"max_drawdown",
|
|
)
|
|
)
|
|
|
|
if not frames:
|
|
return pl.DataFrame()
|
|
|
|
result = pl.concat(frames, how="diagonal_relaxed")
|
|
# A signature-matched carrier should be one row per cost level, but guard
|
|
# against any residual duplicate deterministically (keep best Sharpe).
|
|
result = result.sort(["case_study", "cost_bps", "sharpe"]).unique(
|
|
subset=["case_study", "cost_bps"], keep="last", maintain_order=True
|
|
)
|
|
name_df = pl.DataFrame([{"case_study": k, "display_name": v} for k, v in SHORT_NAMES.items()])
|
|
return (
|
|
result.join(name_df, on="case_study", how="left")
|
|
.with_columns(cadence=pl.col("case_study").replace(CADENCE_MAP))
|
|
.sort("case_study", "cost_bps")
|
|
)
|