#!/usr/bin/env python3 """Read-only attribution for four already-generated Freqtrade backtests. This tool never runs a backtest. It compares: * ``baseline_a``: production-rule control before the candidate * ``noop``: candidate plumbing with its gate disabled * ``candidate``: the registered V2 pass-through/rank rule * ``baseline_b_repeat``: the unchanged production-rule control rerun afterward Freqtrade result JSON does not contain rejected raw signals, unfilled orders, or the full point-in-time stop/portfolio state. Episodes emitted here are therefore ``RESULT_PATH_DIVERGENCE`` diagnostics based on observable trades. They MUST NOT be counted as formal signal-level ``PATH_DIVERGENCE`` episodes. """ from __future__ import annotations import argparse import csv import hashlib import json import math import sys import zipfile from collections import defaultdict from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterable SCRIPT_VERSION = "1.0" SENSITIVITY_PAIRS = ("EDGE/USDT:USDT", "POWER/USDT:USDT") ARM_NAMES = ("baseline_a", "noop", "candidate", "baseline_b_repeat") FORMAL_CONTRAST = ( "baseline_a_vs_candidate", "baseline_a", "candidate", "registered_v2_effect", ) ISOLATION_ARMS = ("noop", "baseline_b_repeat") PROTOCOL_FIELDS = ( "backtest_start", "backtest_end", "timeframe", "starting_balance", "max_open_trades", "trading_mode", "margin_mode", "enable_protections", ) REQUIRED_RESULT_FIELDS = ( "trades", "profit_total_abs", "profit_total", "profit_factor", "max_drawdown_account", "starting_balance", "final_balance", "backtest_start", "backtest_end", "timeframe", ) TRADE_REQUIRED_FIELDS = ( "pair", "is_short", "open_timestamp", "close_timestamp", "profit_abs", "profit_ratio", ) @dataclass(frozen=True) class LoadedArm: name: str path: Path sha256: str strategy: str result_member: str | None wallet_member: str | None wallet_sha256: str | None result: dict[str, Any] def sha256_file(path: Path, chunk_size: int = 1024 * 1024) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(chunk_size), b""): digest.update(chunk) return digest.hexdigest() def json_payload( path: Path, ) -> tuple[dict[str, Any], str | None, str | None, str | None]: if path.suffix.lower() != ".zip": return json.loads(path.read_text(encoding="utf-8")), None, None, None with zipfile.ZipFile(path) as archive: candidates = [ name for name in archive.namelist() if name.endswith(".json") and not name.endswith("_config.json") and not name.endswith(".meta.json") ] if len(candidates) != 1: raise ValueError( f"{path}: expected exactly one result JSON in zip, found {len(candidates)}" ) wallet_candidates = [ name for name in archive.namelist() if name.endswith("_wallet.feather") ] if len(wallet_candidates) > 1: raise ValueError( f"{path}: expected at most one wallet Feather, found " f"{len(wallet_candidates)}" ) wallet_member = wallet_candidates[0] if wallet_candidates else None wallet_sha = ( hashlib.sha256(archive.read(wallet_member)).hexdigest() if wallet_member is not None else None ) return ( json.loads(archive.read(candidates[0])), candidates[0], wallet_member, wallet_sha, ) def select_strategy(payload: dict[str, Any], requested: str | None, path: Path) -> tuple[str, dict[str, Any]]: strategies = payload.get("strategy") if isinstance(strategies, dict): if requested is not None: if requested not in strategies: raise ValueError( f"{path}: strategy {requested!r} not found; choices={sorted(strategies)}" ) return requested, strategies[requested] if len(strategies) != 1: raise ValueError( f"{path}: contains {len(strategies)} strategies; specify the arm's --*-strategy" ) name, result = next(iter(strategies.items())) return str(name), result # A bare strategy result is accepted only with an explicit strategy label. if requested is None: raise ValueError(f"{path}: bare result JSON requires an explicit strategy name") return requested, payload def as_float(value: Any, label: str) -> float: try: result = float(value) except (TypeError, ValueError) as exc: raise ValueError(f"{label} must be numeric") from exc if not math.isfinite(result): raise ValueError(f"{label} must be finite") return result def validate_result(result: dict[str, Any], path: Path) -> None: missing = [field for field in REQUIRED_RESULT_FIELDS if field not in result] if missing: raise ValueError(f"{path}: result missing fields: {', '.join(missing)}") if not isinstance(result["trades"], list): raise ValueError(f"{path}: trades must be a list") for field in ( "profit_total_abs", "profit_total", "profit_factor", "max_drawdown_account", "starting_balance", "final_balance", ): as_float(result[field], f"{path}: {field}") for index, trade in enumerate(result["trades"]): missing_trade = [field for field in TRADE_REQUIRED_FIELDS if field not in trade] if missing_trade: raise ValueError( f"{path}: trade {index} missing fields: {', '.join(missing_trade)}" ) as_float(trade["profit_abs"], f"{path}: trade {index} profit_abs") as_float(trade["profit_ratio"], f"{path}: trade {index} profit_ratio") if trade["close_timestamp"] is None: raise ValueError(f"{path}: trade {index} has no close_timestamp") trade_profit = sum(float(trade["profit_abs"]) for trade in result["trades"]) if not math.isclose( trade_profit, float(result["profit_total_abs"]), rel_tol=1e-10, abs_tol=1e-8, ): raise ValueError( f"{path}: profit_total_abs does not equal summed trade profit " f"({result['profit_total_abs']} != {trade_profit})" ) def load_arm(name: str, path: Path, strategy: str | None) -> LoadedArm: if not path.is_file(): raise ValueError(f"{name} input does not exist: {path}") payload, member, wallet_member, wallet_sha = json_payload(path) strategy_name, result = select_strategy(payload, strategy, path) validate_result(result, path) return LoadedArm( name=name, path=path, sha256=sha256_file(path), strategy=strategy_name, result_member=member, wallet_member=wallet_member, wallet_sha256=wallet_sha, result=result, ) def side_name(trade: dict[str, Any]) -> str: return "short" if bool(trade["is_short"]) else "long" def entry_key_tuple(trade: dict[str, Any]) -> tuple[str, str, int, str]: return ( str(trade["pair"]), side_name(trade), int(trade["open_timestamp"]), str(trade.get("enter_tag") or ""), ) def entry_key_text(key: tuple[str, str, int, str]) -> str: return "|".join((key[0], key[1], str(key[2]), key[3])) def trade_fingerprint(trade: dict[str, Any]) -> tuple[Any, ...]: orders = trade.get("orders") or [] order_fingerprint = tuple( ( order.get("ft_order_side"), order.get("ft_is_entry"), order.get("order_filled_timestamp"), order.get("safe_price"), order.get("amount"), ) for order in orders ) return ( entry_key_tuple(trade), int(trade["close_timestamp"]), trade.get("exit_reason"), trade.get("open_rate"), trade.get("close_rate"), trade.get("amount"), trade.get("stake_amount"), trade.get("leverage"), trade.get("initial_stop_loss_abs"), trade.get("stop_loss_abs"), float(trade["profit_abs"]), float(trade["profit_ratio"]), trade.get("funding_fees"), order_fingerprint, ) def trade_map(result: dict[str, Any], arm: str) -> dict[tuple[str, str, int, str], dict[str, Any]]: mapped: dict[tuple[str, str, int, str], dict[str, Any]] = {} for trade in result["trades"]: key = entry_key_tuple(trade) if key in mapped: raise ValueError( f"{arm}: duplicate trade entry key {entry_key_text(key)}; " "safe matching is ambiguous" ) mapped[key] = trade return mapped def profit_factor(trades: Iterable[dict[str, Any]]) -> float | None: values = [float(trade["profit_abs"]) for trade in trades] gains = sum(value for value in values if value > 0) losses = -sum(value for value in values if value < 0) if losses == 0: return None return gains / losses def realized_path(result: dict[str, Any]) -> list[dict[str, Any]]: by_time: dict[int, float] = defaultdict(float) for trade in result["trades"]: by_time[int(trade["close_timestamp"])] += float(trade["profit_abs"]) balance = float(result["starting_balance"]) output = [ { "timestamp": int(datetime.fromisoformat( str(result["backtest_start"]).replace("Z", "+00:00") ).replace(tzinfo=timezone.utc).timestamp() * 1000), "realized_equity": balance, "realized_pnl": 0.0, } ] for timestamp, pnl in sorted(by_time.items()): balance += pnl output.append( { "timestamp": timestamp, "realized_equity": balance, "realized_pnl": pnl, } ) return output def aligned_realized_paths(arms: dict[str, LoadedArm]) -> list[dict[str, Any]]: paths = {name: realized_path(arm.result) for name, arm in arms.items()} times = sorted( { int(row["timestamp"]) for path in paths.values() for row in path } ) output = [] for timestamp in times: row: dict[str, Any] = { "timestamp": timestamp, "time": iso_timestamp(timestamp), "path_kind": "realized_equity_at_trade_close_boundaries", } for arm_name, path in paths.items(): row[f"{arm_name}_realized_equity"] = equity_at(path, timestamp) row[f"{arm_name}_event_profit"] = sum( float(item["realized_pnl"]) for item in path if int(item["timestamp"]) == timestamp ) output.append(row) return output def realized_drawdown(path: list[dict[str, Any]]) -> dict[str, float]: peak = float(path[0]["realized_equity"]) max_abs = 0.0 max_ratio = 0.0 for row in path: equity = float(row["realized_equity"]) peak = max(peak, equity) drawdown = peak - equity ratio = drawdown / peak if peak else 0.0 max_abs = max(max_abs, drawdown) max_ratio = max(max_ratio, ratio) return {"absolute": max_abs, "ratio": max_ratio} def arm_metrics(arm: LoadedArm) -> dict[str, Any]: result = arm.result trades = result["trades"] path = realized_path(result) return { "strategy": arm.strategy, "archive_sha256": arm.sha256, "result_member": arm.result_member, "wallet_member": arm.wallet_member, "wallet_sha256": arm.wallet_sha256, "trades": len(trades), "profit_abs": float(result["profit_total_abs"]), "profit_ratio": float(result["profit_total"]), "profit_factor_exported": float(result["profit_factor"]), "profit_factor_recomputed": profit_factor(trades), "max_drawdown_ratio_exported": float(result["max_drawdown_account"]), "realized_close_event_drawdown": realized_drawdown(path), "starting_balance": float(result["starting_balance"]), "final_balance": float(result["final_balance"]), "long_trades": sum(not bool(trade["is_short"]) for trade in trades), "short_trades": sum(bool(trade["is_short"]) for trade in trades), "force_exit_trades": sum(trade.get("exit_reason") == "force_exit" for trade in trades), } def compare_trades(left: LoadedArm, right: LoadedArm) -> tuple[dict[str, Any], list[dict[str, Any]]]: left_map = trade_map(left.result, left.name) right_map = trade_map(right.result, right.name) left_keys = set(left_map) right_keys = set(right_map) common = left_keys & right_keys exact = {key for key in common if trade_fingerprint(left_map[key]) == trade_fingerprint(right_map[key])} modified = common - exact differences: list[dict[str, Any]] = [] for key in sorted(left_keys | right_keys): left_trade = left_map.get(key) right_trade = right_map.get(key) if key in exact: classification = "identical" elif left_trade is not None and right_trade is not None: classification = "modified" elif left_trade is not None: classification = "left_only" else: classification = "right_only" differences.append( { "entry_key": entry_key_text(key), "pair": key[0], "side": key[1], "open_timestamp": key[2], "classification": classification, "left_close_timestamp": ( int(left_trade["close_timestamp"]) if left_trade is not None else None ), "right_close_timestamp": ( int(right_trade["close_timestamp"]) if right_trade is not None else None ), "left_exit_reason": left_trade.get("exit_reason") if left_trade else None, "right_exit_reason": right_trade.get("exit_reason") if right_trade else None, "left_profit_abs": ( float(left_trade["profit_abs"]) if left_trade is not None else 0.0 ), "right_profit_abs": ( float(right_trade["profit_abs"]) if right_trade is not None else 0.0 ), "profit_delta": ( (float(right_trade["profit_abs"]) if right_trade is not None else 0.0) - (float(left_trade["profit_abs"]) if left_trade is not None else 0.0) ), "_left_trade": left_trade, "_right_trade": right_trade, } ) left_count = len(left_map) return { "left_trades": left_count, "right_trades": len(right_map), "matched_entry_trades": len(common), "identical_trades": len(exact), "modified_trades": len(modified), "left_only_trades": len(left_keys - right_keys), "right_only_trades": len(right_keys - left_keys), "entry_retention_ratio": len(common) / left_count if left_count else None, "exact_trade_retention_ratio": len(exact) / left_count if left_count else None, }, differences def difference_interval(row: dict[str, Any]) -> tuple[int, int]: left = row["_left_trade"] right = row["_right_trade"] opens = [ int(trade["open_timestamp"]) for trade in (left, right) if trade is not None ] closes = [ int(trade["close_timestamp"]) for trade in (left, right) if trade is not None ] if row["classification"] in {"left_only", "right_only"}: return min(opens), max(closes) assert left is not None and right is not None open_fields = ( "open_rate", "amount", "stake_amount", "leverage", "initial_stop_loss_abs", ) if any(left.get(field) != right.get(field) for field in open_fields): return min(opens), max(closes) if int(left["close_timestamp"]) != int(right["close_timestamp"]): return min(int(left["close_timestamp"]), int(right["close_timestamp"])), max(closes) # Same active-position interval but a different exit fill/reason/PnL. close_time = int(left["close_timestamp"]) return close_time, close_time def equity_at(path: list[dict[str, Any]], timestamp: int) -> float: value = float(path[0]["realized_equity"]) for row in path[1:]: if int(row["timestamp"]) > timestamp: break value = float(row["realized_equity"]) return value def episode_max_equity_divergence( left_path: list[dict[str, Any]], right_path: list[dict[str, Any]], start: int, end: int, ) -> float: times = {start, end} times.update(int(row["timestamp"]) for row in left_path if start <= int(row["timestamp"]) <= end) times.update(int(row["timestamp"]) for row in right_path if start <= int(row["timestamp"]) <= end) return max(abs(equity_at(right_path, stamp) - equity_at(left_path, stamp)) for stamp in times) def divergence_episodes( left: LoadedArm, right: LoadedArm, differences: list[dict[str, Any]], ) -> list[dict[str, Any]]: divergent = [row for row in differences if row["classification"] != "identical"] intervals = sorted( [(difference_interval(row), row) for row in divergent], key=lambda item: (item[0][0], item[0][1], item[1]["entry_key"]), ) groups: list[tuple[int, int, list[dict[str, Any]]]] = [] for (start, end), row in intervals: if groups and start <= groups[-1][1]: prior_start, prior_end, rows = groups[-1] groups[-1] = (prior_start, max(prior_end, end), [*rows, row]) else: groups.append((start, end, [row])) left_path = realized_path(left.result) right_path = realized_path(right.result) output = [] for index, (start, end, rows) in enumerate(groups, 1): left_pnl = sum(float(row["left_profit_abs"]) for row in rows) right_pnl = sum(float(row["right_profit_abs"]) for row in rows) censored = any( trade is not None and trade.get("exit_reason") == "force_exit" for row in rows for trade in (row["_left_trade"], row["_right_trade"]) ) output.append( { "episode_id": f"{left.name}_vs_{right.name}-{index:04d}", "classification": "RESULT_PATH_DIVERGENCE", "formal_path_divergence_eligible": False, "start_timestamp": start, "end_timestamp": end, "censored": censored, "affected_trade_count": len(rows), "affected_pairs": sorted({row["pair"] for row in rows}), "affected_sides": sorted({row["side"] for row in rows}), "difference_classes": dict( sorted( { kind: sum(row["classification"] == kind for row in rows) for kind in ("modified", "left_only", "right_only") }.items() ) ), "entry_keys": [row["entry_key"] for row in rows], "left_attributed_profit_abs": left_pnl, "right_attributed_profit_abs": right_pnl, "attributed_profit_delta": right_pnl - left_pnl, "max_realized_equity_divergence": episode_max_equity_divergence( left_path, right_path, start, end ), "evidence_limit": ( "Result trades only; no rejected signals, unfilled orders, " "dynamic stop state, or mark-to-market equity." ), } ) return output def arithmetic_sensitivity( left: LoadedArm, right: LoadedArm, ) -> list[dict[str, Any]]: scenarios = ( ("none", frozenset()), ("exclude_EDGE", frozenset((SENSITIVITY_PAIRS[0],))), ("exclude_POWER", frozenset((SENSITIVITY_PAIRS[1],))), ("exclude_EDGE_POWER", frozenset(SENSITIVITY_PAIRS)), ) rows = [] for label, excluded in scenarios: left_trades = [trade for trade in left.result["trades"] if trade["pair"] not in excluded] right_trades = [trade for trade in right.result["trades"] if trade["pair"] not in excluded] left_pnl = sum(float(trade["profit_abs"]) for trade in left_trades) right_pnl = sum(float(trade["profit_abs"]) for trade in right_trades) rows.append( { "scenario": label, "excluded_pairs": sorted(excluded), "left_trades": len(left_trades), "right_trades": len(right_trades), "left_profit_abs": left_pnl, "right_profit_abs": right_pnl, "profit_delta": right_pnl - left_pnl, "left_profit_factor": profit_factor(left_trades), "right_profit_factor": profit_factor(right_trades), "method": "arithmetic_trade_removal_without_portfolio_rerun", "drawdown_not_recomputed": True, } ) return rows def protocol_check(arms: dict[str, LoadedArm]) -> dict[str, Any]: baseline = arms["baseline_a"].result differences: list[dict[str, Any]] = [] for arm_name in ARM_NAMES[1:]: result = arms[arm_name].result for field in PROTOCOL_FIELDS: if result.get(field) != baseline.get(field): differences.append( { "arm": arm_name, "field": field, "baseline": baseline.get(field), "observed": result.get(field), } ) baseline_fees = sorted( { (trade.get("fee_open"), trade.get("fee_close")) for trade in baseline["trades"] } ) arm_fees = sorted( {(trade.get("fee_open"), trade.get("fee_close")) for trade in result["trades"]} ) if arm_fees != baseline_fees: differences.append( { "arm": arm_name, "field": "trade_fee_pairs", "baseline": baseline_fees, "observed": arm_fees, } ) return {"valid": not differences, "differences": differences} def public_trade_row(row: dict[str, Any]) -> dict[str, Any]: return {key: value for key, value in row.items() if not key.startswith("_")} def iso_timestamp(milliseconds: int) -> str: return datetime.fromtimestamp(milliseconds / 1000, tz=timezone.utc).isoformat() def write_csv(path: Path, rows: list[dict[str, Any]]) -> None: fields = sorted({key for row in rows for key in row}) if rows else ("empty",) with path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fields) writer.writeheader() for row in rows: writer.writerow( { key: ( json.dumps(value, ensure_ascii=False, sort_keys=True) if isinstance(value, (dict, list)) else value ) for key, value in row.items() } ) def build_report(arms: dict[str, LoadedArm]) -> tuple[dict[str, Any], dict[str, list[dict[str, Any]]]]: protocol = protocol_check(arms) metrics = {name: arm_metrics(arm) for name, arm in arms.items()} isolation_metric_fields = ( "trades", "profit_abs", "profit_ratio", "profit_factor_exported", "max_drawdown_ratio_exported", "starting_balance", "final_balance", ) isolation_checks: dict[str, Any] = {} for arm_name in ISOLATION_ARMS: retention, differences = compare_trades(arms["baseline_a"], arms[arm_name]) metrics_identical = all( metrics["baseline_a"][field] == metrics[arm_name][field] for field in isolation_metric_fields ) normalized_trades_identical = ( retention["left_only_trades"] == 0 and retention["right_only_trades"] == 0 and retention["modified_trades"] == 0 ) wallet_path_available = ( arms["baseline_a"].wallet_sha256 is not None and arms[arm_name].wallet_sha256 is not None ) wallet_path_identical = ( wallet_path_available and arms["baseline_a"].wallet_sha256 == arms[arm_name].wallet_sha256 ) isolation_checks[f"baseline_a_vs_{arm_name}"] = { "normalized_trades_identical": normalized_trades_identical, "metrics_identical": metrics_identical, "wallet_path_available": wallet_path_available, "wallet_path_identical": wallet_path_identical, "trade_comparison": retention, "unexpected_trade_differences": [ public_trade_row(row) for row in differences if row["classification"] != "identical" ], "passed": ( normalized_trades_identical and metrics_identical and wallet_path_identical ), } sandwich_isolation_valid = all( check["passed"] for check in isolation_checks.values() ) contrast_name, left_name, right_name, purpose = FORMAL_CONTRAST left = arms[left_name] right = arms[right_name] retention, differences = compare_trades(left, right) episodes = divergence_episodes(left, right, differences) sensitivity = arithmetic_sensitivity(left, right) formal_contrast = { "contrast": contrast_name, "left_arm": left_name, "right_arm": right_name, "purpose": purpose, "trade_comparison": retention, "metric_delta": { "trades": metrics[right_name]["trades"] - metrics[left_name]["trades"], "profit_abs": metrics[right_name]["profit_abs"] - metrics[left_name]["profit_abs"], "profit_ratio": metrics[right_name]["profit_ratio"] - metrics[left_name]["profit_ratio"], "profit_factor": ( metrics[right_name]["profit_factor_exported"] - metrics[left_name]["profit_factor_exported"] ), "max_drawdown_ratio": ( metrics[right_name]["max_drawdown_ratio_exported"] - metrics[left_name]["max_drawdown_ratio_exported"] ), }, "result_path_episode_count": len(episodes), "completed_result_path_episode_count": sum(not row["censored"] for row in episodes), "episodes": episodes, "edge_power_arithmetic_sensitivity": sensitivity, } report = { "script_version": SCRIPT_VERSION, "generated_at": datetime.now(timezone.utc).isoformat(), "episode_contract": { "classification": "RESULT_PATH_DIVERGENCE", "formal_path_divergence_eligible": False, "reason": ( "Freqtrade result archives omit rejected raw signals, unfilled " "orders, and full active-state history." ), }, "arms": { name: { "path": str(arm.path.resolve()), **metrics[name], } for name, arm in arms.items() }, "protocol": protocol, "sandwich_isolation": { "required_equality": "baseline_a == noop == baseline_b_repeat", "checks": isolation_checks, "passed": sandwich_isolation_valid, }, "formal_contrast": formal_contrast, "summary": { "protocol_valid": protocol["valid"], "sandwich_isolation_valid": sandwich_isolation_valid, "safe_to_interpret_candidate_contrast": ( protocol["valid"] and sandwich_isolation_valid ), }, } tables = { "metrics": [{"arm": name, **row} for name, row in metrics.items()], "trade_differences": [ {"contrast": contrast_name, **public_trade_row(row)} for row in differences if row["classification"] != "identical" ], "divergence_episodes": [ { "contrast": contrast_name, **row, "start_time": iso_timestamp(row["start_timestamp"]), "end_time": iso_timestamp(row["end_timestamp"]), } for row in episodes ], "edge_power_sensitivity": [ {"contrast": contrast_name, **row} for row in sensitivity ], "realized_equity_path": aligned_realized_paths(arms), } return report, tables def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) for arm in ARM_NAMES: option = arm.replace("_", "-") parser.add_argument(f"--{option}", type=Path, required=True) parser.add_argument(f"--{option}-strategy") parser.add_argument("--output-dir", type=Path, required=True) return parser def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) try: arms = { arm: load_arm( arm, getattr(args, arm), getattr(args, f"{arm}_strategy"), ) for arm in ARM_NAMES } report, tables = build_report(arms) except (ValueError, json.JSONDecodeError, zipfile.BadZipFile) as exc: raise SystemExit(str(exc)) from exc args.output_dir.mkdir(parents=True, exist_ok=True) json_path = args.output_dir / "paired_backtest_attribution.json" json_path.write_text( json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True, allow_nan=False) + "\n", encoding="utf-8", ) for name, rows in tables.items(): write_csv(args.output_dir / f"{name}.csv", rows) print(json.dumps(report["summary"], ensure_ascii=False, sort_keys=True)) print(f"wrote {json_path}") return 0 if report["summary"]["safe_to_interpret_candidate_contrast"] else 1 if __name__ == "__main__": raise SystemExit(main())