mirror of
https://github.com/NicolasBohn/NexQuant.git
synced 2026-07-29 08:27:43 +00:00
feat: unified backtest engine, LLM error handling, strategy refactor
- Add vbt_backtest.py as single source of truth for all metric formulas (Sharpe, drawdown, IC, transaction costs) — backtest_engine.py and strategy_orchestrator.py now delegate to it - Add LLMUnavailableError to exception.py; rd_loop.py catches it at the proposal stage and raises LoopResumeError to avoid corrupting trace history with None hypotheses - Guard record() against None exp/hypothesis so loop resets leave trace.hist in a consistent state - Refactor strategy_orchestrator and optuna_optimizer to use unified backtest path; remove duplicate metric calculation code - Add predix_rebacktest_unified.py script for offline re-evaluation - Update tests and README Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
TPTBusiness
@@ -25,9 +25,19 @@ from rich.console import Console
|
||||
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TimeElapsedColumn
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# Suppress warnings
|
||||
# Suppress warnings and noisy loggers that bleed into Rich progress output
|
||||
warnings.filterwarnings('ignore')
|
||||
logging.getLogger('rdagent').setLevel(logging.WARNING)
|
||||
for _noisy in ('rdagent', 'litellm', 'LiteLLM', 'litellm.utils',
|
||||
'litellm.main', 'httpx', 'httpcore', 'openai', 'urllib3'):
|
||||
logging.getLogger(_noisy).setLevel(logging.CRITICAL)
|
||||
# Suppress litellm verbose flag if already imported
|
||||
try:
|
||||
import litellm as _ll
|
||||
_ll.suppress_debug_info = True
|
||||
_ll.verbose = False
|
||||
_ll.set_verbose = False
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# ============================================================================
|
||||
# Configuration
|
||||
@@ -54,10 +64,12 @@ else:
|
||||
MIN_IC = 0.02
|
||||
MIN_SHARPE = 0.5
|
||||
MIN_TRADES = 10
|
||||
MAX_DRAWDOWN = -1.0
|
||||
MAX_DRAWDOWN = -0.30
|
||||
STYLE_EMOJI = '📈 Swing'
|
||||
STYLE_DESC = 'medium-term intraday'
|
||||
|
||||
TXN_COST_BPS = float(os.getenv('TXN_COST_BPS', '1.0'))
|
||||
|
||||
console = Console()
|
||||
|
||||
# ============================================================================
|
||||
@@ -65,10 +77,10 @@ console = Console()
|
||||
# ============================================================================
|
||||
def setup_llm_env():
|
||||
"""Setup LLM environment variables."""
|
||||
load_dotenv(Path(__file__).parent / '.env')
|
||||
router_key = os.getenv('OPENROUTER_API_KEY') or os.getenv('OPENAI_API_KEY', '')
|
||||
if not router_key or router_key == 'local':
|
||||
router_key = os.getenv('OPENROUTER_API_KEY', '')
|
||||
load_dotenv(Path(__file__).parent.parent / '.env')
|
||||
if os.getenv('OPENAI_API_KEY') == 'local' or os.getenv('LLM_BACKEND', '').lower() == 'local':
|
||||
return
|
||||
router_key = os.getenv('OPENROUTER_API_KEY', '')
|
||||
if router_key:
|
||||
os.environ['OPENAI_API_KEY'] = router_key
|
||||
os.environ['OPENAI_API_BASE'] = 'https://openrouter.ai/api/v1'
|
||||
@@ -211,121 +223,80 @@ Output ONLY valid JSON with these fields:
|
||||
# Backtest Runner (runs in main process to avoid re-loading data)
|
||||
# ============================================================================
|
||||
def run_backtest(close, factors_df, strategy_code):
|
||||
"""Run real backtest with actual OHLCV data."""
|
||||
"""
|
||||
Execute LLM-generated strategy code in a sandboxed subprocess to produce
|
||||
the signal, then delegate all metric computation to the unified
|
||||
``backtest_signal`` engine in the main process.
|
||||
"""
|
||||
if close is None or factors_df is None or len(factors_df.columns) < 2:
|
||||
return None
|
||||
|
||||
|
||||
import tempfile
|
||||
|
||||
|
||||
# Subprocess stays minimal: it only runs the untrusted strategy code
|
||||
# and pickles the resulting signal. All numbers come from the shared engine.
|
||||
script = f"""
|
||||
import pandas as pd
|
||||
import numpy as np
|
||||
import json
|
||||
|
||||
close = pd.read_pickle('close.pkl')
|
||||
factors = pd.read_pickle('factors.pkl')
|
||||
|
||||
try:
|
||||
{chr(10).join(' ' + l for l in strategy_code.split(chr(10)))}
|
||||
except:
|
||||
print("ERROR: Strategy execution failed")
|
||||
exit(1)
|
||||
except Exception as e:
|
||||
print(f"ERROR: Strategy execution failed: {{e}}")
|
||||
raise SystemExit(1)
|
||||
|
||||
if 'signal' not in dir():
|
||||
print("ERROR: No signal generated")
|
||||
exit(1)
|
||||
raise SystemExit(1)
|
||||
|
||||
signal = signal.fillna(0)
|
||||
common_idx = close.index.intersection(signal.index)
|
||||
close = close.loc[common_idx]
|
||||
signal = signal.loc[common_idx]
|
||||
|
||||
FORWARD_BARS = {FORWARD_BARS}
|
||||
returns_fwd = close.pct_change(FORWARD_BARS).shift(-FORWARD_BARS)
|
||||
signal_aligned = signal.loc[returns_fwd.dropna().index]
|
||||
fwd_returns = returns_fwd.loc[signal_aligned.index]
|
||||
|
||||
if len(signal_aligned) < 100 or len(fwd_returns) < 100:
|
||||
print("ERROR: Not enough data after alignment")
|
||||
exit(1)
|
||||
|
||||
ic = signal_aligned.corr(fwd_returns)
|
||||
strategy_returns = signal_aligned * fwd_returns
|
||||
|
||||
if strategy_returns.std() > 0:
|
||||
sharpe = strategy_returns.mean() / strategy_returns.std() * np.sqrt(252 * 1440 / {FORWARD_BARS})
|
||||
else:
|
||||
sharpe = 0
|
||||
|
||||
cum = (1 + strategy_returns).cumprod()
|
||||
running_max = cum.expanding().max()
|
||||
drawdown = (cum - running_max) / running_max.replace(0, np.nan)
|
||||
max_dd = drawdown.min() if len(drawdown) > 0 else 0
|
||||
|
||||
win_rate = (strategy_returns > 0).sum() / len(strategy_returns) if len(strategy_returns) > 0 else 0
|
||||
n_trades = int((signal_aligned != signal_aligned.shift(1)).sum())
|
||||
|
||||
total_return = cum.iloc[-1] - 1
|
||||
n_bars = len(strategy_returns)
|
||||
n_months = n_bars / (252 * 1440 / {FORWARD_BARS} / 12) if n_bars > 0 else 1
|
||||
|
||||
if n_months > 0 and (1 + total_return) > 0:
|
||||
monthly_return = (1 + total_return) ** (1 / n_months) - 1
|
||||
annual_return = (1 + total_return) ** (12 / n_months) - 1
|
||||
else:
|
||||
monthly_return = total_return
|
||||
annual_return = total_return * 12
|
||||
|
||||
result = {{
|
||||
"status": "success",
|
||||
"sharpe": float(sharpe),
|
||||
"max_drawdown": float(max_dd) if not np.isnan(max_dd) else -0.20,
|
||||
"win_rate": float(win_rate),
|
||||
"ic": float(ic) if not np.isnan(ic) else 0,
|
||||
"n_trades": n_trades,
|
||||
"total_return": float(total_return),
|
||||
"monthly_return_pct": float(monthly_return * 100),
|
||||
"annual_return_pct": float(annual_return * 100),
|
||||
"n_bars": int(n_bars),
|
||||
"n_months": float(n_months),
|
||||
"signal_long": int((signal_aligned == 1).sum()),
|
||||
"signal_short": int((signal_aligned == -1).sum()),
|
||||
"signal_neutral": int((signal_aligned == 0).sum()),
|
||||
}}
|
||||
|
||||
print(json.dumps(result))
|
||||
signal.fillna(0).to_pickle('signal.pkl')
|
||||
"""
|
||||
|
||||
|
||||
with tempfile.TemporaryDirectory() as td:
|
||||
tdp = Path(td)
|
||||
close.to_pickle(str(tdp / 'close.pkl'))
|
||||
factors_df.to_pickle(str(tdp / 'factors.pkl'))
|
||||
|
||||
script_path = tdp / 'run.py'
|
||||
script_path.write_text(script)
|
||||
|
||||
(tdp / 'run.py').write_text(script)
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
['python', str(script_path)],
|
||||
['python', 'run.py'],
|
||||
capture_output=True, text=True, timeout=60,
|
||||
cwd=str(tdp)
|
||||
)
|
||||
|
||||
if result.returncode != 0:
|
||||
return {'status': 'failed', 'reason': result.stderr[:200] or result.stdout[:200]}
|
||||
|
||||
for line in result.stdout.strip().split('\n'):
|
||||
try:
|
||||
return json.loads(line)
|
||||
except:
|
||||
continue
|
||||
|
||||
return {'status': 'failed', 'reason': 'No valid JSON output'}
|
||||
return {'status': 'failed', 'reason': (result.stderr or result.stdout)[:200]}
|
||||
|
||||
signal = pd.read_pickle(tdp / 'signal.pkl')
|
||||
except subprocess.TimeoutExpired:
|
||||
return {'status': 'failed', 'reason': 'Timeout (60s)'}
|
||||
except Exception as e:
|
||||
return {'status': 'failed', 'reason': str(e)[:200]}
|
||||
|
||||
# Main process: unified backtest (identical formulas everywhere).
|
||||
from rdagent.components.backtesting.vbt_backtest import backtest_signal
|
||||
|
||||
common = close.index.intersection(signal.index)
|
||||
if len(common) < 100:
|
||||
return {'status': 'failed', 'reason': f'Not enough aligned data ({len(common)} bars)'}
|
||||
|
||||
close_a = close.loc[common]
|
||||
signal_a = signal.reindex(common).fillna(0)
|
||||
|
||||
# Forward returns at the configured horizon feed IC computation.
|
||||
fwd_returns = close_a.pct_change(FORWARD_BARS).shift(-FORWARD_BARS)
|
||||
|
||||
return backtest_signal(
|
||||
close=close_a,
|
||||
signal=signal_a,
|
||||
txn_cost_bps=TXN_COST_BPS,
|
||||
freq='1min',
|
||||
forward_returns=fwd_returns,
|
||||
)
|
||||
|
||||
# ============================================================================
|
||||
# Main Parallel Strategy Generation
|
||||
# ============================================================================
|
||||
@@ -393,6 +364,8 @@ def main(target_count=10):
|
||||
BarColumn(),
|
||||
TextColumn("[bold green]{task.completed}/{task.total}"),
|
||||
TimeElapsedColumn(),
|
||||
redirect_stdout=True,
|
||||
redirect_stderr=True,
|
||||
) as progress:
|
||||
task = progress.add_task("Generating...", total=max_attempts)
|
||||
|
||||
|
||||
@@ -0,0 +1,280 @@
|
||||
#!/usr/bin/env python
|
||||
"""
|
||||
Re-run existing strategies through the unified backtest engine.
|
||||
|
||||
For every strategy JSON in results/strategies_new (or a user-supplied dir):
|
||||
1. Load the factor values it references.
|
||||
2. Execute its ``code`` in a sandboxed subprocess to produce the signal.
|
||||
3. Run the signal through ``backtest_signal`` on REAL 1-min EUR/USD close.
|
||||
4. Print old-vs-new sharpe / DD / trades / total-return so the impact of
|
||||
the unified engine (no return clipping, proper 1-min annualization,
|
||||
trade-epoch win rate) is visible.
|
||||
|
||||
Does NOT mutate the strategy JSON files — read-only comparison.
|
||||
|
||||
Usage:
|
||||
python scripts/predix_rebacktest_unified.py # all strategies
|
||||
python scripts/predix_rebacktest_unified.py 50 # first 50
|
||||
python scripts/predix_rebacktest_unified.py 50 --csv report.csv
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
from rich.console import Console
|
||||
from rich.progress import BarColumn, Progress, SpinnerColumn, TextColumn, TimeElapsedColumn
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
from rdagent.components.backtesting.vbt_backtest import backtest_signal # noqa: E402
|
||||
|
||||
OHLCV_PATH = Path("/home/nico/Predix/git_ignore_folder/factor_implementation_source_data/intraday_pv.h5")
|
||||
FACTORS_VALUES_DIR = Path("/home/nico/Predix/results/factors/values")
|
||||
STRATEGIES_DIR = Path("/home/nico/Predix/results/strategies_new")
|
||||
|
||||
console = Console()
|
||||
|
||||
|
||||
def load_close() -> pd.Series:
|
||||
ohlcv = pd.read_hdf(str(OHLCV_PATH), key="data")
|
||||
col = "$close" if "$close" in ohlcv.columns else "close"
|
||||
close = ohlcv[col].dropna()
|
||||
# Drop the "EURUSD" instrument level if present — the strategies work
|
||||
# on a single series indexed by timestamp.
|
||||
if isinstance(close.index, pd.MultiIndex):
|
||||
close = close.droplevel(-1)
|
||||
return close.astype(float).sort_index()
|
||||
|
||||
|
||||
def load_factor_series(names: List[str]) -> Dict[str, pd.Series]:
|
||||
out: Dict[str, pd.Series] = {}
|
||||
for name in names:
|
||||
for variant in (name, name.replace("/", "_").replace("\\", "_")[:150]):
|
||||
path = FACTORS_VALUES_DIR / f"{variant}.parquet"
|
||||
if path.exists():
|
||||
try:
|
||||
df = pd.read_parquet(str(path))
|
||||
if df is not None and len(df.columns) > 0:
|
||||
out[name] = df.iloc[:, 0]
|
||||
break
|
||||
except Exception:
|
||||
pass
|
||||
return out
|
||||
|
||||
|
||||
def execute_strategy(
|
||||
factors_df: pd.DataFrame,
|
||||
close: pd.Series,
|
||||
strategy_code: str,
|
||||
timeout: int = 45,
|
||||
) -> Optional[pd.Series]:
|
||||
"""Run untrusted LLM code in a subprocess and return the resulting signal."""
|
||||
script = f"""
|
||||
import pandas as pd, numpy as np
|
||||
factors = pd.read_pickle('factors.pkl')
|
||||
close = pd.read_pickle('close.pkl')
|
||||
df = factors # some strategies reference 'df', others 'factors'
|
||||
|
||||
try:
|
||||
{chr(10).join(' ' + line for line in strategy_code.split(chr(10)))}
|
||||
except Exception as e:
|
||||
print(f"ERROR: {{e}}")
|
||||
raise SystemExit(1)
|
||||
|
||||
if 'signal' not in dir():
|
||||
print("ERROR: no signal")
|
||||
raise SystemExit(1)
|
||||
|
||||
pd.Series(signal).fillna(0).to_pickle('signal.pkl')
|
||||
"""
|
||||
with tempfile.TemporaryDirectory() as td:
|
||||
tdp = Path(td)
|
||||
factors_df.to_pickle(str(tdp / "factors.pkl"))
|
||||
close.to_pickle(str(tdp / "close.pkl"))
|
||||
(tdp / "run.py").write_text(script)
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["python", "run.py"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=timeout,
|
||||
cwd=str(tdp),
|
||||
)
|
||||
if result.returncode != 0:
|
||||
return None
|
||||
signal = pd.read_pickle(tdp / "signal.pkl")
|
||||
return signal
|
||||
except (subprocess.TimeoutExpired, Exception):
|
||||
return None
|
||||
|
||||
|
||||
def rebacktest_one(
|
||||
strategy_data: Dict[str, Any],
|
||||
close: pd.Series,
|
||||
txn_cost_bps: float,
|
||||
) -> Dict[str, Any]:
|
||||
factor_names = strategy_data.get("factor_names") or strategy_data.get("factors_used") or []
|
||||
code = strategy_data.get("code", "")
|
||||
if not factor_names or not code:
|
||||
return {"status": "skipped", "reason": "missing factors or code"}
|
||||
|
||||
factor_series = load_factor_series(factor_names)
|
||||
if len(factor_series) < 2:
|
||||
return {"status": "skipped", "reason": f"only {len(factor_series)} factor files found"}
|
||||
|
||||
factors_df = pd.DataFrame(factor_series).dropna(how="all")
|
||||
if isinstance(factors_df.index, pd.MultiIndex):
|
||||
factors_df = factors_df.droplevel(-1)
|
||||
factors_df = factors_df.sort_index()
|
||||
|
||||
# Factors are typically daily-timestamped; close is 1-min.
|
||||
# Direct index intersection would be near-zero → reindex and ffill first,
|
||||
# matching exactly what the orchestrator's evaluate_strategy does.
|
||||
factors_1min = factors_df.reindex(close.index).ffill()
|
||||
valid_rows = factors_1min.notna().any(axis=1)
|
||||
if valid_rows.sum() < 1000:
|
||||
return {"status": "skipped", "reason": f"only {valid_rows.sum()} valid rows after ffill"}
|
||||
|
||||
close_a = close.loc[valid_rows]
|
||||
factors_a = factors_1min.loc[valid_rows]
|
||||
|
||||
signal = execute_strategy(factors_a, close_a, code)
|
||||
if signal is None:
|
||||
return {"status": "code_failed"}
|
||||
|
||||
# Signal can arrive on either the factor index or the close index.
|
||||
signal = signal.reindex(close_a.index).ffill().fillna(0)
|
||||
|
||||
result = backtest_signal(
|
||||
close=close_a,
|
||||
signal=signal,
|
||||
txn_cost_bps=txn_cost_bps,
|
||||
freq="1min",
|
||||
)
|
||||
result["status_detail"] = result.pop("status")
|
||||
result["status"] = "ok"
|
||||
return result
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("count", type=int, nargs="?", default=None,
|
||||
help="Limit to first N strategies (default: all)")
|
||||
parser.add_argument("--dir", type=Path, default=STRATEGIES_DIR,
|
||||
help="Strategy directory to re-backtest")
|
||||
parser.add_argument("--csv", type=Path, default=None,
|
||||
help="Write a CSV report to this path")
|
||||
parser.add_argument("--txn-cost-bps", type=float, default=1.5)
|
||||
args = parser.parse_args()
|
||||
|
||||
console.print(f"[cyan]Loading OHLCV close...[/cyan]")
|
||||
close = load_close()
|
||||
console.print(f"[green]✓[/green] {len(close):,} 1-min bars "
|
||||
f"({close.index[0]} → {close.index[-1]})\n")
|
||||
|
||||
files = sorted(args.dir.glob("*.json"))
|
||||
if args.count:
|
||||
files = files[:args.count]
|
||||
console.print(f"[cyan]Re-backtesting {len(files)} strategies with unified engine...[/cyan]\n")
|
||||
|
||||
rows: List[Dict[str, Any]] = []
|
||||
with Progress(
|
||||
SpinnerColumn(),
|
||||
TextColumn("[bold blue]{task.description}"),
|
||||
BarColumn(),
|
||||
TextColumn("[bold green]{task.completed}/{task.total}"),
|
||||
TimeElapsedColumn(),
|
||||
) as progress:
|
||||
task = progress.add_task("Backtesting", total=len(files))
|
||||
for f in files:
|
||||
try:
|
||||
data = json.load(open(f))
|
||||
except Exception:
|
||||
progress.update(task, advance=1)
|
||||
continue
|
||||
|
||||
old = data.get("summary", {})
|
||||
name = data.get("strategy_name", f.stem)[:38]
|
||||
|
||||
bt = rebacktest_one(data, close, args.txn_cost_bps)
|
||||
|
||||
row = {
|
||||
"file": f.name,
|
||||
"name": name,
|
||||
"status": bt.get("status"),
|
||||
"reason": bt.get("reason", bt.get("status_detail", "")),
|
||||
"old_sharpe": old.get("sharpe"),
|
||||
"old_dd": old.get("max_drawdown"),
|
||||
"old_trades": old.get("real_n_trades"),
|
||||
"old_monthly_pct": old.get("monthly_return_pct"),
|
||||
"new_sharpe": bt.get("sharpe"),
|
||||
"new_dd": bt.get("max_drawdown"),
|
||||
"new_trades": bt.get("n_trades"),
|
||||
"new_total_return": bt.get("total_return"),
|
||||
"new_annual_return_cagr": None,
|
||||
"data_quality": bt.get("data_quality_flag"),
|
||||
}
|
||||
# annualized CAGR is not in forward_returns wrapper; use annual_return_pct/100 proxy
|
||||
if "annualized_return" in bt:
|
||||
row["new_annual_return_cagr"] = bt["annualized_return"]
|
||||
rows.append(row)
|
||||
progress.update(task, advance=1)
|
||||
|
||||
# Summary
|
||||
ok_rows = [r for r in rows if r["status"] == "ok"]
|
||||
console.print(f"\n[bold]{len(ok_rows)}/{len(rows)} strategies successfully re-backtested[/bold]\n")
|
||||
|
||||
status_counts: Dict[str, int] = {}
|
||||
for r in rows:
|
||||
status_counts[r["status"]] = status_counts.get(r["status"], 0) + 1
|
||||
for status, n in sorted(status_counts.items(), key=lambda kv: -kv[1]):
|
||||
console.print(f" {status}: {n}")
|
||||
|
||||
if ok_rows:
|
||||
# Compare old vs new where both exist
|
||||
comparable = [r for r in ok_rows if r["old_sharpe"] is not None]
|
||||
if comparable:
|
||||
old_sharpe = np.array([r["old_sharpe"] for r in comparable], dtype=float)
|
||||
new_sharpe = np.array([r["new_sharpe"] for r in comparable], dtype=float)
|
||||
console.print(f"\n[bold]Sharpe drift ({len(comparable)} strategies with old metrics):[/bold]")
|
||||
console.print(f" old mean={old_sharpe.mean():+.3f} median={np.median(old_sharpe):+.3f} max={old_sharpe.max():+.3f}")
|
||||
console.print(f" new mean={new_sharpe.mean():+.3f} median={np.median(new_sharpe):+.3f} max={new_sharpe.max():+.3f}")
|
||||
diff = new_sharpe - old_sharpe
|
||||
console.print(f" Δ mean={diff.mean():+.3f} median={np.median(diff):+.3f}")
|
||||
agree_sign = int(((np.sign(old_sharpe) == np.sign(new_sharpe)) | (np.abs(new_sharpe) < 0.1)).sum())
|
||||
console.print(f" sign-agreement: {agree_sign}/{len(comparable)} "
|
||||
f"({agree_sign/len(comparable):.0%})")
|
||||
|
||||
ok_rows.sort(key=lambda r: r["new_sharpe"] if r["new_sharpe"] is not None else -1e9, reverse=True)
|
||||
console.print(f"\n[bold]Top 15 by new Sharpe:[/bold]")
|
||||
console.print(f" {'name':<38} {'old_sh':>7} {'new_sh':>7} {'new_dd':>8} {'new_trd':>7} {'new_ret':>9}")
|
||||
for r in ok_rows[:15]:
|
||||
osh = f"{r['old_sharpe']:+.2f}" if r["old_sharpe"] is not None else " —"
|
||||
ddv = f"{r['new_dd']:.2%}" if r["new_dd"] is not None else "—"
|
||||
rtv = f"{r['new_total_return']:+.2%}" if r["new_total_return"] is not None else "—"
|
||||
console.print(f" {r['name']:<38} {osh:>7} {r['new_sharpe']:>+7.2f} {ddv:>8} {r['new_trades'] or 0:>7} {rtv:>9}")
|
||||
|
||||
flagged = [r for r in ok_rows if r["data_quality"]]
|
||||
if flagged:
|
||||
console.print(f"\n[yellow]⚠ {len(flagged)} strategies flagged with extreme bars "
|
||||
f"(would have been hidden by old ±10% clipping)[/yellow]")
|
||||
|
||||
if args.csv:
|
||||
with open(args.csv, "w", newline="") as fh:
|
||||
w = csv.DictWriter(fh, fieldnames=list(rows[0].keys()))
|
||||
w.writeheader()
|
||||
w.writerows(rows)
|
||||
console.print(f"\n[green]✓[/green] CSV report written to {args.csv}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user