Skip to content

Tidy quantitative validation and replay code #4

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
ReviveCoding wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Tidy quantitative validation and replay code #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions scripts/evaluate_performance.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,12 +1,11 @@
from __future__ import annotations
from __future__ import annotations

import json
import subprocess
import sys
import time
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
OUTPUTS = ROOT / "outputs"
REPORTS = ROOT / "reports"
Expand Down
66 changes: 49 additions & 17 deletions scripts/run_ablation_study.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,9 @@
from __future__ import annotations
from __future__ import annotations

from pathlib import Path

import pandas as pd


ROOT = Path(__file__).resolve().parents[1]
OUTPUTS = ROOT / "outputs"
REPORTS = ROOT / "reports"
Expand Down Expand Up @@ -49,11 +48,15 @@ def main() -> None:

high_or_critical_findings = 0
if len(findings) and "severity" in findings.columns:
high_or_critical_findings = int(findings["severity"].isin(["HIGH", "CRITICAL"]).sum())
high_or_critical_findings = int(
findings["severity"].isin(["HIGH", "CRITICAL"]).sum()
)

residual_mae_reduction = None
if len(residual_summary):
possible_cols = [c for c in residual_summary.columns if "reduction" in c.lower()]
possible_cols = [
c for c in residual_summary.columns if "reduction" in c.lower()
]
if possible_cols:
residual_mae_reduction = float(residual_summary[possible_cols[0]].iloc[0])

Expand All @@ -77,7 +80,9 @@ def main() -> None:
if "pass_rate" in operational.columns:
operational_pass_rate = float(operational["pass_rate"].iloc[0])
if "recalibration_trigger_rate" in operational.columns:
recalibration_trigger_rate = float(operational["recalibration_trigger_rate"].iloc[0])
recalibration_trigger_rate = float(
operational["recalibration_trigger_rate"].iloc[0]
)

ablation = pd.DataFrame(
[
Expand All @@ -93,26 +98,41 @@ def main() -> None:
{
"framework_version": "+ Benchmark and numerical reliability",
"key_added_layer": "Closed-form benchmarks, Greeks checks, convergence, seed stability",
"false_pass_rate": round(max(final_false_pass, baseline_false_pass * 0.40), 4),
"false_pass_reduction_vs_baseline_pct": pct_reduction(baseline_false_pass, max(final_false_pass, baseline_false_pass * 0.40)),
"false_pass_rate": round(
max(final_false_pass, baseline_false_pass * 0.40), 4
),
"false_pass_reduction_vs_baseline_pct": pct_reduction(
baseline_false_pass,
max(final_false_pass, baseline_false_pass * 0.40),
),
"validation_health_score": None,
"high_or_critical_findings": None,
"resume_message": "Improves detection of numerical instability and benchmark mismatch.",
},
{
"framework_version": "+ Asymmetric validation envelope",
"key_added_layer": "Risk-underestimation penalties and severity-aware validation",
"false_pass_rate": round(max(final_false_pass, baseline_false_pass * 0.15), 4),
"false_pass_reduction_vs_baseline_pct": pct_reduction(baseline_false_pass, max(final_false_pass, baseline_false_pass * 0.15)),
"false_pass_rate": round(
max(final_false_pass, baseline_false_pass * 0.15), 4
),
"false_pass_reduction_vs_baseline_pct": pct_reduction(
baseline_false_pass,
max(final_false_pass, baseline_false_pass * 0.15),
),
"validation_health_score": None,
"high_or_critical_findings": high_or_critical_findings,
"resume_message": "Prioritizes dangerous VaR/stress underestimation over symmetric error.",
},
{
"framework_version": "+ Residual correction challenger",
"key_added_layer": "Residual pricing/risk error correction and champion/challenger comparison",
"false_pass_rate": round(max(final_false_pass, baseline_false_pass * 0.05), 4),
"false_pass_reduction_vs_baseline_pct": pct_reduction(baseline_false_pass, max(final_false_pass, baseline_false_pass * 0.05)),
"false_pass_rate": round(
max(final_false_pass, baseline_false_pass * 0.05), 4
),
"false_pass_reduction_vs_baseline_pct": pct_reduction(
baseline_false_pass,
max(final_false_pass, baseline_false_pass * 0.05),
),
"validation_health_score": None,
"high_or_critical_findings": high_or_critical_findings,
"resume_message": "Adds challenger evidence for explainable residual-error reduction.",
Expand All @@ -121,7 +141,9 @@ def main() -> None:
"framework_version": "Final validation framework",
"key_added_layer": "Backtesting, APL/HPL proxy, operational replay, runbook, release gate",
"false_pass_rate": final_false_pass,
"false_pass_reduction_vs_baseline_pct": pct_reduction(baseline_false_pass, final_false_pass),
"false_pass_reduction_vs_baseline_pct": pct_reduction(
baseline_false_pass, final_false_pass
),
"validation_health_score": validation_health_score,
"high_or_critical_findings": high_or_critical_findings,
"resume_message": "End-to-end validation workflow reduces false-pass risk and supports operational monitoring.",
Expand All @@ -148,17 +170,25 @@ def main() -> None:
]

if residual_mae_reduction is not None:
report_lines.append(f"- Residual correction improvement metric: {residual_mae_reduction:.4f}")
report_lines.append(
f"- Residual correction improvement metric: {residual_mae_reduction:.4f}"
)
if pnl_correlation is not None:
report_lines.append(f"- P&L attribution proxy correlation: {pnl_correlation:.4f}")
report_lines.append(
f"- P&L attribution proxy correlation: {pnl_correlation:.4f}"
)
if recent_250_exceptions is not None:
report_lines.append(f"- Recent-250 VaR exceptions: {recent_250_exceptions}")
if traffic_light_zone is not None:
report_lines.append(f"- Backtesting traffic-light zone: {traffic_light_zone}")
if operational_pass_rate is not None:
report_lines.append(f"- Operational replay pass rate: {operational_pass_rate:.4f}")
report_lines.append(
f"- Operational replay pass rate: {operational_pass_rate:.4f}"
)
if recalibration_trigger_rate is not None:
report_lines.append(f"- Operational replay recalibration-trigger rate: {recalibration_trigger_rate:.4f}")
report_lines.append(
f"- Operational replay recalibration-trigger rate: {recalibration_trigger_rate:.4f}"
)

report_lines.extend(
[
Expand All @@ -174,7 +204,9 @@ def main() -> None:
]
)

(REPORTS / "baseline_vs_enhanced_ablation_report.md").write_text("\n".join(report_lines), encoding="utf-8")
(REPORTS / "baseline_vs_enhanced_ablation_report.md").write_text(
"\n".join(report_lines), encoding="utf-8"
)

print("Ablation study completed.")
print(ablation.to_string(index=False))
Expand Down
55 changes: 42 additions & 13 deletions scripts/run_operational_replay.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
from __future__ import annotations
from __future__ import annotations

import json
import subprocess
Expand All @@ -9,7 +9,6 @@
import numpy as np
import pandas as pd


ROOT = Path(__file__).resolve().parents[1]
OUTPUTS = ROOT / "outputs"
REPORTS = ROOT / "reports"
Expand All @@ -30,13 +29,25 @@ def run_pipeline_once(run_id: int, operating_day: str) -> dict:
findings_path = OUTPUTS / "model_risk_findings.csv"
backtest_path = OUTPUTS / "backtesting_summary.csv"

release = json.loads(release_path.read_text(encoding="utf-8")) if release_path.exists() else {}
release = (
json.loads(release_path.read_text(encoding="utf-8"))
if release_path.exists()
else {}
)
health = pd.read_csv(health_path) if health_path.exists() else pd.DataFrame()
findings = pd.read_csv(findings_path) if findings_path.exists() else pd.DataFrame()
backtest = pd.read_csv(backtest_path) if backtest_path.exists() else pd.DataFrame()

health_score = float(health["validation_health_score"].iloc[0]) if "validation_health_score" in health.columns and len(health) else np.nan
final_false_pass = float(health["false_pass_rate"].iloc[0]) if "false_pass_rate" in health.columns and len(health) else np.nan
health_score = (
float(health["validation_health_score"].iloc[0])
if "validation_health_score" in health.columns and len(health)
else np.nan
)
final_false_pass = (
float(health["false_pass_rate"].iloc[0])
if "false_pass_rate" in health.columns and len(health)
else np.nan
)

high_or_critical = 0
if len(findings) and "severity" in findings.columns:
Expand Down Expand Up @@ -83,13 +94,23 @@ def build_recalibration_summary(log: pd.DataFrame) -> pd.DataFrame:
{
"num_operating_runs": len(log),
"pass_rate": round((log["release_decision"] == "PASS").mean(), 4),
"review_or_block_rate": round(log["release_decision"].isin(["REVIEW", "BLOCK"]).mean(), 4),
"recalibration_trigger_rate": round(log["recalibration_trigger"].mean(), 4),
"review_or_block_rate": round(
log["release_decision"].isin(["REVIEW", "BLOCK"]).mean(), 4
),
"recalibration_trigger_rate": round(
log["recalibration_trigger"].mean(), 4
),
"avg_runtime_seconds": round(log["runtime_seconds"].mean(), 4),
"max_runtime_seconds": round(log["runtime_seconds"].max(), 4),
"avg_validation_health_score": round(log["validation_health_score"].mean(), 4),
"min_validation_health_score": round(log["validation_health_score"].min(), 4),
"max_recent_250_exceptions": round(log["recent_250_exceptions"].max(), 4),
"avg_validation_health_score": round(
log["validation_health_score"].mean(), 4
),
"min_validation_health_score": round(
log["validation_health_score"].min(), 4
),
"max_recent_250_exceptions": round(
log["recent_250_exceptions"].max(), 4
),
}
]
)
Expand Down Expand Up @@ -131,7 +152,9 @@ def write_reports(log: pd.DataFrame, summary: pd.DataFrame) -> None:
"- This is not a regulatory capital engine and should be described as a synthetic/public-sample operational replay.",
]

(REPORTS / "operational_replay_report.md").write_text("\n".join(report), encoding="utf-8")
(REPORTS / "operational_replay_report.md").write_text(
"\n".join(report), encoding="utf-8"
)

runbook = [
"# Model Operations Runbook",
Expand Down Expand Up @@ -161,7 +184,9 @@ def write_reports(log: pd.DataFrame, summary: pd.DataFrame) -> None:
"This runbook supports a local, synthetic/public-sample, non-production validation project. It does not claim production trading, official regulatory capital, or bank-approved model use.",
]

(REPORTS / "model_operations_runbook.md").write_text("\n".join(runbook), encoding="utf-8")
(REPORTS / "model_operations_runbook.md").write_text(
"\n".join(runbook), encoding="utf-8"
)


def main() -> None:
Expand All @@ -181,7 +206,11 @@ def main() -> None:
summary.to_csv(OUTPUTS / "recalibration_summary.csv", index=False)
write_reports(log, summary)

status = "PASS" if (log["return_code"].eq(0).all() and summary["pass_rate"].iloc[0] >= 0.8) else "REVIEW"
status = (
"PASS"
if (log["return_code"].eq(0).all() and summary["pass_rate"].iloc[0] >= 0.8)
else "REVIEW"
)

print("Operational replay completed.")
print(f"Status: {status}")
Expand Down
Loading
Loading