from __future__ import annotations import argparse import json from dataclasses import dataclass from pathlib import Path from statistics import mean from typing import Mapping, Sequence from config import BotConfig, RuntimePaths from historical_replay import Candle, ReplayResult, ReplayTrade @dataclass(frozen=True) class LabStrategy: strategy_id: str family: str parameters: dict[str, float | int | str | bool] @dataclass(frozen=True) class OpenLabPosition: coin: str entry_ts: float entry_price: float size: float entry_index: int family: str def _param(strategy: LabStrategy, key: str, default: float | int | str | bool) -> float | int | str | bool: return strategy.parameters.get(key, default) def _pct_change(current: float, reference: float) -> float: if reference <= 0: return 0.0 return (current - reference) / reference * 100.0 def _rsi(closes: Sequence[float], period: int = 14) -> float: if len(closes) < period + 1: return 50.0 gains: list[float] = [] losses: list[float] = [] for prev, cur in zip(closes[-period - 1 : -1], closes[-period:]): delta = cur - prev if delta >= 0: gains.append(delta) losses.append(0.0) else: gains.append(0.0) losses.append(abs(delta)) avg_gain = mean(gains) if gains else 0.0 avg_loss = mean(losses) if losses else 0.0 if avg_loss == 0: return 100.0 if avg_gain > 0 else 50.0 rs = avg_gain / avg_loss return 100.0 - (100.0 / (1.0 + rs)) def _sma(values: Sequence[float], period: int) -> float: if not values: return 0.0 window = values[-max(1, period) :] return mean(window) def _std(values: Sequence[float]) -> float: if len(values) < 2: return 0.0 avg = mean(values) return (sum((item - avg) ** 2 for item in values) / len(values)) ** 0.5 def _atr_pct(candles: Sequence[Candle], idx: int, period: int) -> float | None: if idx <= 0 or idx < period: return None true_ranges: list[float] = [] start = idx - period + 1 for cur_idx in range(start, idx + 1): candle = candles[cur_idx] previous_close = candles[cur_idx - 1].close true_ranges.append(max(candle.high - candle.low, abs(candle.high - previous_close), abs(candle.low - previous_close))) close = candles[idx].close if close <= 0: return None return mean(true_ranges) / close * 100.0 def _relative_strength_pct(candles: Sequence[Candle], idx: int, lookback: int) -> float | None: if idx < lookback: return None reference = candles[idx - lookback].close return _pct_change(candles[idx].close, reference) def _is_top_relative_strength( strategy: LabStrategy, coin: str, candles_by_coin: Mapping[str, Sequence[Candle]], idx: int, ) -> bool: lookback = int(_param(strategy, "lookback", 24)) top_n = max(1, int(_param(strategy, "top_n", 3))) min_momentum = float(_param(strategy, "min_momentum_pct", 2.0)) min_positive_candidates = max(1, int(_param(strategy, "min_positive_candidates", 1))) scores: list[tuple[float, str]] = [] for candidate, rows in candles_by_coin.items(): if idx >= len(rows): continue score = _relative_strength_pct(rows, idx, lookback) if score is not None: scores.append((score, candidate.upper())) eligible = [(score, candidate) for score, candidate in scores if score >= min_momentum] if len(eligible) < min_positive_candidates: return False eligible.sort(key=lambda item: (item[0], item[1]), reverse=True) top = {candidate for _score, candidate in eligible[:top_n]} return coin.upper() in top def _should_enter(strategy: LabStrategy, candles: Sequence[Candle], idx: int) -> bool: if idx <= 0: return False family = strategy.family close = candles[idx].close closes = [c.close for c in candles[: idx + 1]] previous = candles[:idx] if family == "momentum_breakout": lookback = int(_param(strategy, "lookback", 20)) if len(previous) < lookback: return False breakout_pct = float(_param(strategy, "breakout_pct", 1.0)) recent_high = max(c.high for c in previous[-lookback:]) fast = int(_param(strategy, "fast_sma", max(2, lookback // 2))) slow = int(_param(strategy, "slow_sma", lookback)) trend_ok = _sma(closes[:-1], fast) >= _sma(closes[:-1], slow) return trend_ok and _pct_change(close, recent_high) >= breakout_pct if family == "rsi_mean_reversion": period = int(_param(strategy, "rsi_period", 14)) oversold = float(_param(strategy, "oversold_rsi", 30.0)) return _rsi(closes, period) <= oversold if family == "trend_pullback": fast = int(_param(strategy, "fast_sma", 8)) slow = int(_param(strategy, "slow_sma", 21)) lookback = int(_param(strategy, "lookback", slow)) pullback_pct = float(_param(strategy, "pullback_pct", 2.0)) if len(previous) < max(fast, slow, lookback): return False recent_high = max(c.high for c in previous[-lookback:]) bounced = close > candles[idx - 1].close return _sma(closes[:-1], fast) > _sma(closes[:-1], slow) and _pct_change(close, recent_high) <= -pullback_pct and bounced if family == "volatility_squeeze_breakout": lookback = int(_param(strategy, "lookback", 20)) if len(previous) < lookback: return False basis_values = [c.close for c in previous[-lookback:]] basis = mean(basis_values) width_pct = (_std(basis_values) * 4.0 / basis * 100.0) if basis else 999.0 max_width = float(_param(strategy, "max_band_width_pct", 4.0)) breakout_pct = float(_param(strategy, "breakout_pct", 0.75)) return width_pct <= max_width and _pct_change(close, max(c.high for c in previous[-lookback:])) >= breakout_pct if family == "liquidation_reversal": lookback = int(_param(strategy, "lookback", 12)) if len(previous) < lookback: return False recent_high = max(c.high for c in previous[-lookback:]) drop_pct = abs(float(_param(strategy, "drop_pct", 3.0))) if _pct_change(close, recent_high) > -drop_pct: return False prior_volumes = [c.volume for c in previous[-lookback:] if c.volume > 0] if not prior_volumes or candles[idx].volume <= 0: return False volume_spike_mult = float(_param(strategy, "volume_spike_mult", 2.0)) return candles[idx].volume >= mean(prior_volumes) * volume_spike_mult if family == "risk_managed_trend_following": fast = int(_param(strategy, "fast_sma", 12)) slow = int(_param(strategy, "slow_sma", 36)) long_lookback = int(_param(strategy, "long_lookback", slow * 2)) if len(previous) < max(fast, slow, long_lookback): return False fast_ok = _sma(closes[:-1], fast) > _sma(closes[:-1], slow) long_trend = _pct_change(close, closes[-long_lookback - 1]) min_long_trend = float(_param(strategy, "min_long_trend_pct", 3.0)) atr_period = int(_param(strategy, "atr_period", 14)) max_atr_pct = float(_param(strategy, "max_atr_pct", 999.0)) atr_pct = _atr_pct(candles, idx, atr_period) volatility_ok = atr_pct is None or atr_pct <= max_atr_pct return fast_ok and long_trend >= min_long_trend and volatility_ok if family == "donchian_volume_breakout": lookback = int(_param(strategy, "lookback", 20)) if len(previous) < lookback: return False recent_high = max(c.high for c in previous[-lookback:]) breakout_pct = float(_param(strategy, "breakout_pct", 0.5)) if _pct_change(close, recent_high) < breakout_pct: return False prior_volumes = [c.volume for c in previous[-lookback:] if c.volume > 0] if not prior_volumes or candles[idx].volume <= 0: return False volume_spike_mult = float(_param(strategy, "volume_spike_mult", 1.5)) return candles[idx].volume >= mean(prior_volumes) * volume_spike_mult return False def _exit_reason(strategy: LabStrategy, position: OpenLabPosition, candles: Sequence[Candle], idx: int) -> str | None: close = candles[idx].close pnl_pct = _pct_change(close, position.entry_price) stop_loss = float(_param(strategy, "stop_loss_pct", 3.0)) take_profit = float(_param(strategy, "take_profit_pct", 2.0)) max_hold = int(_param(strategy, "max_hold_bars", 24)) if pnl_pct <= -abs(stop_loss): return "stop_loss" if pnl_pct >= abs(take_profit): return "take_profit" if strategy.family == "rsi_mean_reversion": period = int(_param(strategy, "rsi_period", 14)) exit_rsi = float(_param(strategy, "exit_rsi", 55.0)) if _rsi([c.close for c in candles[: idx + 1]], period) >= exit_rsi and pnl_pct > 0: return "rsi_recovered" if idx - position.entry_index >= max_hold: return "time_stop" return None def _max_drawdown(pnls: Sequence[float]) -> float: equity = 0.0 peak = 0.0 max_dd = 0.0 for pnl in pnls: equity += pnl peak = max(peak, equity) max_dd = min(max_dd, equity - peak) return round(max_dd, 8) def _profit_factor(pnls: Sequence[float]) -> float: gains = sum(p for p in pnls if p > 0) losses = abs(sum(p for p in pnls if p < 0)) if losses == 0: return round(gains, 8) if gains else 0.0 return round(gains / losses, 8) def _score(pnls: Sequence[float]) -> float: closed = len(pnls) if not pnls: return -2.0 avg_pnl = mean(pnls) win_rate = sum(1 for p in pnls if p > 0) / closed pf = min(_profit_factor(pnls), 5.0) dd_penalty = abs(_max_drawdown(pnls)) * 0.35 sample_penalty = 1.0 if closed < 5 else 0.0 return round(avg_pnl + win_rate * 0.5 + pf * 0.1 - dd_penalty - sample_penalty, 8) def _close_trade(strategy: LabStrategy, position: OpenLabPosition, candle: Candle, reason: str) -> ReplayTrade: pnl = (candle.close - position.entry_price) * position.size return ReplayTrade( strategy_id=strategy.strategy_id, coin=position.coin, entry_ts=position.entry_ts, exit_ts=float(candle.ts), entry_price=position.entry_price, exit_price=candle.close, size=position.size, pnl_usd=round(pnl, 8), exit_reason=reason, minutes_in_trade=int((float(candle.ts) - position.entry_ts) // 60), ) def run_lab_strategy(strategy: LabStrategy, candles_by_coin: Mapping[str, Sequence[Candle]], *, volumes: Mapping[str, float]) -> ReplayResult: min_volume = float(_param(strategy, "min_volume_24h", 1.0)) trade_size = float(_param(strategy, "trade_size_usd", 50.0)) max_positions = int(_param(strategy, "max_total_trades", 3)) open_positions: dict[str, OpenLabPosition] = {} trades: list[ReplayTrade] = [] timeline: list[tuple[float, str, Candle, Sequence[Candle], int]] = [] for coin, candles in candles_by_coin.items(): rows = list(candles) for idx, candle in enumerate(rows): timeline.append((float(candle.ts), coin.upper(), candle, rows, idx)) timeline.sort(key=lambda item: item[0]) for _ts, coin, candle, coin_candles, idx in timeline: if volumes.get(coin, 0.0) < min_volume: continue if coin in open_positions: reason = _exit_reason(strategy, open_positions[coin], coin_candles, idx) if reason: trades.append(_close_trade(strategy, open_positions.pop(coin), candle, reason)) continue if len(open_positions) >= max_positions: continue if strategy.family == "relative_strength_rotation": enter = _is_top_relative_strength(strategy, coin, candles_by_coin, idx) else: enter = _should_enter(strategy, coin_candles, idx) if enter: open_positions[coin] = OpenLabPosition( coin=coin, entry_ts=float(candle.ts), entry_price=candle.close, size=trade_size / candle.close, entry_index=idx, family=strategy.family, ) last_by_coin = {coin.upper(): list(candles)[-1] for coin, candles in candles_by_coin.items() if candles} for coin, position in list(open_positions.items()): last = last_by_coin.get(coin) if last is not None: trades.append(_close_trade(strategy, position, last, "end_of_replay")) pnls = [trade.pnl_usd for trade in trades] closed = len(pnls) return ReplayResult( strategy_id=strategy.strategy_id, closed_trades=closed, win_rate=round(sum(1 for pnl in pnls if pnl > 0) / closed, 4) if closed else 0.0, avg_pnl_usd=round(mean(pnls), 8) if pnls else 0.0, total_pnl_usd=round(sum(pnls), 8), max_drawdown_usd=_max_drawdown(pnls), profit_factor=_profit_factor(pnls), score=_score(pnls), trades=trades, ) def run_strategy_lab(strategies: Sequence[LabStrategy], candles_by_coin: Mapping[str, Sequence[Candle]], *, volumes: Mapping[str, float]) -> list[ReplayResult]: results = [run_lab_strategy(strategy, candles_by_coin, volumes=volumes) for strategy in strategies] return sorted(results, key=lambda result: (result.score, result.total_pnl_usd, result.closed_trades), reverse=True) def render_strategy_lab_report(results: Sequence[ReplayResult], *, min_samples: int = 20, sampler_strategies: Sequence[LabStrategy] | None = None) -> str: lines = ["Strategy Lab Tournament", "Mode: side-effect-free archetype replay"] if sampler_strategies: lines.append("Research sampler — not a champion candidate") for sampler in sampler_strategies: label = str(sampler.parameters.get("label", sampler.strategy_id)) lines.append(f"- {sampler.strategy_id}: {label}; paper-only research-sampler data-collection") for idx, result in enumerate(results, start=1): warning = " observation-only" if result.closed_trades < min_samples else "" coins = sorted({trade.coin for trade in result.trades}) lines.append( f"{idx}. {result.strategy_id}: score={result.score:+.4f}, n={result.closed_trades}, " f"win={result.win_rate:.2f}, avg={result.avg_pnl_usd:+.4f}, total={result.total_pnl_usd:+.4f}, " f"max_dd={result.max_drawdown_usd:+.4f}, pf={result.profit_factor:.2f}, coins={','.join(coins[:8]) or '-'}{warning}" ) return "\n".join(lines) def default_lab_strategies() -> list[LabStrategy]: return [ LabStrategy("lab_momentum_breakout_fast", "momentum_breakout", {"lookback": 12, "breakout_pct": 0.9, "take_profit_pct": 1.8, "stop_loss_pct": 1.2, "max_hold_bars": 18, "trade_size_usd": 50.0, "min_volume_24h": 10_000_000}), LabStrategy("lab_momentum_breakout_slow", "momentum_breakout", {"lookback": 36, "breakout_pct": 1.6, "take_profit_pct": 3.0, "stop_loss_pct": 1.8, "max_hold_bars": 48, "trade_size_usd": 50.0, "min_volume_24h": 10_000_000}), LabStrategy("lab_rsi_reversion_fast", "rsi_mean_reversion", {"rsi_period": 7, "oversold_rsi": 24.0, "exit_rsi": 55.0, "take_profit_pct": 2.4, "stop_loss_pct": 2.0, "max_hold_bars": 24, "trade_size_usd": 50.0, "min_volume_24h": 10_000_000}), LabStrategy("lab_rsi_reversion_deep", "rsi_mean_reversion", {"rsi_period": 14, "oversold_rsi": 22.0, "exit_rsi": 52.0, "take_profit_pct": 3.5, "stop_loss_pct": 2.5, "max_hold_bars": 36, "trade_size_usd": 50.0, "min_volume_24h": 10_000_000}), LabStrategy("lab_trend_pullback", "trend_pullback", {"fast_sma": 8, "slow_sma": 21, "lookback": 24, "pullback_pct": 2.0, "take_profit_pct": 2.2, "stop_loss_pct": 1.6, "max_hold_bars": 30, "trade_size_usd": 50.0, "min_volume_24h": 10_000_000}), LabStrategy("lab_squeeze_breakout", "volatility_squeeze_breakout", {"lookback": 24, "max_band_width_pct": 3.2, "breakout_pct": 0.7, "take_profit_pct": 2.0, "stop_loss_pct": 1.3, "max_hold_bars": 24, "trade_size_usd": 50.0, "min_volume_24h": 10_000_000}), LabStrategy("lab_liquidation_reversal", "liquidation_reversal", {"lookback": 12, "drop_pct": 3.0, "volume_spike_mult": 2.0, "take_profit_pct": 1.8, "stop_loss_pct": 1.4, "max_hold_bars": 12, "trade_size_usd": 50.0, "min_volume_24h": 10_000_000}), LabStrategy("lab_relative_strength_rotation", "relative_strength_rotation", {"lookback": 36, "top_n": 3, "min_momentum_pct": 3.0, "min_positive_candidates": 3, "take_profit_pct": 3.0, "stop_loss_pct": 1.8, "max_hold_bars": 72, "trade_size_usd": 50.0, "min_volume_24h": 10_000_000}), LabStrategy("lab_risk_managed_trend_following", "risk_managed_trend_following", {"fast_sma": 12, "slow_sma": 36, "long_lookback": 72, "min_long_trend_pct": 4.0, "atr_period": 14, "max_atr_pct": 3.0, "take_profit_pct": 4.0, "stop_loss_pct": 2.0, "max_hold_bars": 96, "trade_size_usd": 50.0, "min_volume_24h": 10_000_000}), LabStrategy("lab_donchian_volume_breakout", "donchian_volume_breakout", {"lookback": 36, "breakout_pct": 0.8, "volume_spike_mult": 1.6, "take_profit_pct": 2.5, "stop_loss_pct": 1.5, "max_hold_bars": 36, "trade_size_usd": 50.0, "min_volume_24h": 10_000_000}), ] def default_research_sampler_strategies() -> list[LabStrategy]: """Paper-only research samplers promoted from recent Strategy-Lab replay. These are explicitly not champion/live candidates; they exist to collect comparable signal/outcome data for promising families without loosening the running v76 candidates. """ return [ LabStrategy( "sampler_squeeze_breakout_research", "volatility_squeeze_breakout", { "label": "Squeeze breakout sampler from strongest replay family", "research_sampler": True, "not_champion": True, "paper_only": True, "data_collection": True, "lookback": 24, "max_band_width_pct": 2.5, "breakout_pct": 0.6, "take_profit_pct": 2.0, "stop_loss_pct": 1.2, "max_hold_bars": 24, "trade_size_usd": 10.0, "max_total_trades": 2, "min_volume_24h": 10_000_000, }, ), LabStrategy( "sampler_relative_strength_research", "relative_strength_rotation", { "label": "Relative strength rotation sampler with breadth guard", "research_sampler": True, "not_champion": True, "paper_only": True, "data_collection": True, "lookback": 24, "top_n": 2, "min_momentum_pct": 3.0, "min_positive_candidates": 3, "take_profit_pct": 3.0, "stop_loss_pct": 1.8, "max_hold_bars": 72, "trade_size_usd": 10.0, "max_total_trades": 2, "min_volume_24h": 10_000_000, }, ), ] def generate_lab_grid_strategies() -> list[LabStrategy]: strategies: list[LabStrategy] = [] for lookback in [8, 16, 32]: for breakout in [0.6, 1.2, 2.0]: safe = str(breakout).replace(".", "p") strategies.append(LabStrategy(f"grid_momo_l{lookback}_b{safe}", "momentum_breakout", {"lookback": lookback, "breakout_pct": breakout, "take_profit_pct": 2.0, "stop_loss_pct": 1.2, "max_hold_bars": 16, "trade_size_usd": 50.0, "min_volume_24h": 10_000_000})) for period in [3, 7, 14]: for oversold in [20, 28, 36]: strategies.append(LabStrategy(f"grid_rsi_p{period}_o{oversold}", "rsi_mean_reversion", {"rsi_period": period, "oversold_rsi": float(oversold), "exit_rsi": 55.0, "take_profit_pct": 2.5, "stop_loss_pct": 1.5, "max_hold_bars": 16, "trade_size_usd": 50.0, "min_volume_24h": 10_000_000})) for pullback in [1.0, 2.5, 4.0]: safe = str(pullback).replace(".", "p") strategies.append(LabStrategy(f"grid_pullback_{safe}", "trend_pullback", {"fast_sma": 8, "slow_sma": 21, "lookback": 24, "pullback_pct": pullback, "take_profit_pct": 2.0, "stop_loss_pct": 1.2, "max_hold_bars": 16, "trade_size_usd": 50.0, "min_volume_24h": 10_000_000})) for width in [2.5, 4.0]: for breakout in [0.6, 1.2]: safe_width = str(width).replace(".", "p") safe_breakout = str(breakout).replace(".", "p") strategies.append(LabStrategy(f"grid_squeeze_w{safe_width}_b{safe_breakout}", "volatility_squeeze_breakout", {"lookback": 24, "max_band_width_pct": width, "breakout_pct": breakout, "take_profit_pct": 2.0, "stop_loss_pct": 1.2, "max_hold_bars": 16, "trade_size_usd": 50.0, "min_volume_24h": 10_000_000})) for drop in [2.5, 3.5, 4.5]: for spike in [1.8, 2.5]: safe_drop = str(drop).replace(".", "p") safe_spike = str(spike).replace(".", "p") strategies.append(LabStrategy(f"grid_liqrev_d{safe_drop}_v{safe_spike}", "liquidation_reversal", {"lookback": 12, "drop_pct": drop, "volume_spike_mult": spike, "take_profit_pct": 1.8, "stop_loss_pct": 1.4, "max_hold_bars": 12, "trade_size_usd": 50.0, "min_volume_24h": 10_000_000})) for lookback in [24, 48]: for top_n in [2, 4]: for min_positive in [1, 3]: strategies.append(LabStrategy(f"grid_rs_l{lookback}_t{top_n}_m{min_positive}", "relative_strength_rotation", {"lookback": lookback, "top_n": top_n, "min_momentum_pct": 3.0, "min_positive_candidates": min_positive, "take_profit_pct": 3.0, "stop_loss_pct": 1.8, "max_hold_bars": 72, "trade_size_usd": 50.0, "min_volume_24h": 10_000_000})) for long_lookback in [48, 96]: for max_atr in [2.5, 4.0]: safe_atr = str(max_atr).replace(".", "p") strategies.append(LabStrategy(f"grid_trend_l{long_lookback}_atr{safe_atr}", "risk_managed_trend_following", {"fast_sma": 12, "slow_sma": 36, "long_lookback": long_lookback, "min_long_trend_pct": 4.0, "atr_period": 14, "max_atr_pct": max_atr, "take_profit_pct": 4.0, "stop_loss_pct": 2.0, "max_hold_bars": 96, "trade_size_usd": 50.0, "min_volume_24h": 10_000_000})) for lookback in [24, 48]: for spike in [1.5, 2.0]: safe_spike = str(spike).replace(".", "p") strategies.append(LabStrategy(f"grid_donchian_l{lookback}_v{safe_spike}", "donchian_volume_breakout", {"lookback": lookback, "breakout_pct": 0.8, "volume_spike_mult": spike, "take_profit_pct": 2.5, "stop_loss_pct": 1.5, "max_hold_bars": 36, "trade_size_usd": 50.0, "min_volume_24h": 10_000_000})) return strategies def _load_candle_cache(cache_dir: Path, coins: Sequence[str], interval: str, hours_back: int) -> dict[str, list[Candle]]: candles: dict[str, list[Candle]] = {} for coin in coins: path = cache_dir / f"{coin.upper()}_{interval}_{hours_back}h.json" if not path.exists(): continue raw = json.loads(path.read_text(encoding="utf-8")) candles[coin.upper()] = [Candle(ts=float(row.get("t", 0)) / 1000.0, open=float(row["o"]), high=float(row["h"]), low=float(row["l"]), close=float(row["c"]), volume=float(row.get("v", 0.0) or 0.0)) for row in raw] return candles def main(argv: list[str] | None = None) -> int: cfg = BotConfig.from_file() paths = RuntimePaths.from_config(cfg) parser = argparse.ArgumentParser(description="Run side-effect-free strategy archetype lab against cached candles.") parser.add_argument("--coins", nargs="*", default=list(cfg.allowed_coins[:16])) parser.add_argument("--interval", default="5m") parser.add_argument("--hours-back", type=int, default=1440) parser.add_argument("--min-samples", type=int, default=20) parser.add_argument("--report-name", default="strategy_lab_report.txt") parser.add_argument("--grid", action="store_true", help="Run the broader archetype parameter grid instead of the small default set.") parser.add_argument("--stride", type=int, default=1, help="Use every Nth candle for faster coarse research runs.") parser.add_argument("--samplers", action="store_true", help="Run/label paper-only research samplers instead of the full archetype set.") args = parser.parse_args(argv) candles = _load_candle_cache(paths.runtime_dir / "candle_cache", args.coins, args.interval, args.hours_back) if args.stride > 1: candles = {coin: rows[:: args.stride] for coin, rows in candles.items()} volumes = {coin: max(sum(c.volume for c in rows), cfg.min_volume_24h) for coin, rows in candles.items()} strategies = default_research_sampler_strategies() if args.samplers else generate_lab_grid_strategies() if args.grid else default_lab_strategies() results = run_strategy_lab(strategies, candles, volumes=volumes) report = render_strategy_lab_report(results, min_samples=args.min_samples, sampler_strategies=strategies if args.samplers else None) paths.reports_dir.mkdir(parents=True, exist_ok=True) report_path = paths.reports_dir / args.report_name report_path.write_text(report, encoding="utf-8") print(report) print(f"Report: {report_path}") return 0 if __name__ == "__main__": raise SystemExit(main())