#!/usr/bin/env python3 """Refresh Binance futures candidate pairs, download data, and backtest. Run this on the deployment host from /data/freqtrade, for example: python3 scripts/refresh_top_pairs_backtest.py --days 60 --max-candidates 30 The default backtest protocol is explicit: 5m, fee 0.00035, cache none. Use ``--candidates-only`` for a read-only liquidity snapshot. Forced incumbent pairs remain in the comparison set but retain their real quote volume and are flagged when they fall below ``--min-quote-volume``. By default the script writes a temporary config and report only. Use --apply to replace the production config whitelist with the selected pairs. """ from __future__ import annotations import argparse import datetime as dt import json import subprocess import sys import urllib.request import zipfile from pathlib import Path from typing import Any BINANCE_FAPI = "https://fapi.binance.com" DEFAULT_TIMEFRAMES = ["5m", "1d"] def utc_yyyymmdd(days_ago: int = 0) -> str: day = dt.datetime.now(dt.timezone.utc).date() - dt.timedelta(days=days_ago) return day.strftime("%Y%m%d") def fetch_json(path: str) -> Any: url = f"{BINANCE_FAPI}{path}" with urllib.request.urlopen(url, timeout=30) as response: return json.load(response) def run(cmd: list[str], cwd: Path) -> None: print("+", " ".join(cmd), flush=True) subprocess.run(cmd, cwd=cwd, check=True) def build_backtest_command( *, service: str, config_arg: str, strategy: str, strategy_path: str, timerange: str, timeframe: str, fee: float, result_dir_arg: str, ) -> list[str]: return [ "docker", "compose", "run", "--rm", service, "backtesting", "--config", config_arg, "--strategy", strategy, "--strategy-path", strategy_path, "--timerange", timerange, "--timeframe", timeframe, "--fee", str(fee), "--export", "trades", "--cache", "none", "--backtest-directory", result_dir_arg, ] def pair_to_symbol(pair: str) -> str: base = pair.split("/")[0] return f"{base}USDT" def load_list(value: str | None) -> set[str]: if not value: return set() path = Path(value) if path.exists(): return { line.strip() for line in path.read_text().splitlines() if line.strip() and not line.strip().startswith("#") } return {item.strip() for item in value.split(",") if item.strip()} def current_top_pairs( *, max_candidates: int, min_quote_volume: float, include: set[str], exclude: set[str], ) -> list[dict[str, Any]]: exchange_info = fetch_json("/fapi/v1/exchangeInfo") tickers = fetch_json("/fapi/v1/ticker/24hr") valid_symbols = { item["symbol"]: item for item in exchange_info["symbols"] if item.get("status") == "TRADING" and item.get("contractType") == "PERPETUAL" and item.get("quoteAsset") == "USDT" } ticker_by_symbol = {ticker.get("symbol"): ticker for ticker in tickers} rows: list[dict[str, Any]] = [] for ticker in tickers: symbol = ticker.get("symbol") if symbol not in valid_symbols: continue quote_volume = float(ticker.get("quoteVolume", 0) or 0) if quote_volume < min_quote_volume: continue base = valid_symbols[symbol]["baseAsset"] pair = f"{base}/USDT:USDT" if pair in exclude or symbol in exclude or base in exclude: continue rows.append( { "symbol": symbol, "pair": pair, "base": base, "quote_volume": quote_volume, } ) rows.sort(key=lambda row: row["quote_volume"], reverse=True) if max_candidates > 0: rows = rows[:max_candidates] existing = {row["pair"] for row in rows} for raw in sorted(include): pair = raw if "/" not in pair: pair = f"{raw}/USDT:USDT" symbol = pair_to_symbol(pair) if pair not in existing and symbol in valid_symbols: ticker = ticker_by_symbol.get(symbol, {}) quote_volume = float(ticker.get("quoteVolume", 0) or 0) rows.append( { "symbol": symbol, "pair": pair, "base": pair.split("/")[0], "quote_volume": quote_volume, "forced": True, "below_min_quote_volume": quote_volume < min_quote_volume, } ) existing.add(pair) return rows def write_config(base_config: Path, output_config: Path, pairs: list[str], bot_name: str) -> None: config = json.loads(base_config.read_text()) config["exchange"]["pair_whitelist"] = pairs config["bot_name"] = bot_name output_config.parent.mkdir(parents=True, exist_ok=True) output_config.write_text(json.dumps(config, indent=4) + "\n") def newest_backtest_zip(directory: Path) -> Path: zips = sorted(directory.glob("backtest-result-*.zip"), key=lambda path: path.stat().st_mtime) if not zips: raise RuntimeError(f"No backtest-result zip found in {directory}") return zips[-1] def load_backtest(zip_path: Path, strategy: str) -> dict[str, Any]: with zipfile.ZipFile(zip_path) as archive: json_names = [ name for name in archive.namelist() if name.endswith(".json") and not name.endswith("_config.json") ] if not json_names: raise RuntimeError(f"No result json found inside {zip_path}") data = json.loads(archive.read(json_names[0])) return data["strategy"][strategy] def select_pairs( result: dict[str, Any], *, min_trades: int, min_profit_pct: float, min_profit_factor: float, max_selected: int, keep: set[str], ) -> list[dict[str, Any]]: selected: list[dict[str, Any]] = [] pair_rows = [row for row in result.get("results_per_pair", []) if row.get("key") != "TOTAL"] for row in pair_rows: pair = row["key"] trades = int(row.get("trades", 0) or 0) profit_pct = float(row.get("profit_total_pct", 0) or 0) profit_factor = row.get("profit_factor") profit_factor = float(profit_factor) if profit_factor is not None else 0.0 forced_keep = pair in keep if not forced_keep: if trades < min_trades: continue if profit_pct < min_profit_pct: continue if profit_factor < min_profit_factor: continue selected.append( { "pair": pair, "trades": trades, "profit_total_pct": profit_pct, "profit_factor": profit_factor, "winrate": float(row.get("winrate", 0) or 0), "max_drawdown_pct": float(row.get("max_drawdown_account", 0) or 0) * 100, "forced_keep": forced_keep, } ) selected.sort( key=lambda row: ( row["forced_keep"], row["profit_total_pct"], row["profit_factor"], row["trades"], ), reverse=True, ) if max_selected > 0: keep_rows = [row for row in selected if row["forced_keep"]] other_rows = [row for row in selected if not row["forced_keep"]] selected = (keep_rows + other_rows)[:max_selected] return selected def apply_whitelist(config_path: Path, pairs: list[str]) -> None: config = json.loads(config_path.read_text()) old = config["exchange"].get("pair_whitelist", []) config["exchange"]["pair_whitelist"] = pairs config_path.write_text(json.dumps(config, indent=4) + "\n") print(f"Applied whitelist to {config_path}") print("Old:", old) print("New:", pairs) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--project-dir", default=".", help="Deployment directory, usually /data/freqtrade.") parser.add_argument("--base-config", default="user_data/config.json") parser.add_argument("--work-config", default="user_data/config-auto-whitelist.json") parser.add_argument("--strategy", default="VolatilityBreakout") parser.add_argument( "--strategy-path", default="user_data/strategies/volatility_breakout", help="Dedicated directory containing the selected strategy family.", ) parser.add_argument("--days", type=int, default=60) parser.add_argument("--start-date", help="Override start date as YYYYMMDD.") parser.add_argument("--end-date", default="", help="Optional end date as YYYYMMDD.") parser.add_argument("--max-candidates", type=int, default=30, help="0 means no cap.") parser.add_argument("--min-quote-volume", type=float, default=100_000_000) parser.add_argument("--timeframes", nargs="+", default=DEFAULT_TIMEFRAMES) parser.add_argument("--timeframe", default="5m", help="Explicit backtest timeframe.") parser.add_argument("--fee", type=float, default=0.00035, help="Backtest fee per side.") parser.add_argument("--include", help="Comma list or file of pairs/assets to force into candidates.") parser.add_argument("--exclude", help="Comma list or file of pairs/assets to exclude from candidates.") parser.add_argument("--keep", help="Comma list or file of pairs to force into selected whitelist.") parser.add_argument("--select-min-trades", type=int, default=10) parser.add_argument("--select-min-profit-pct", type=float, default=0.0) parser.add_argument("--select-min-profit-factor", type=float, default=1.0) parser.add_argument("--select-max-pairs", type=int, default=12, help="0 means no cap.") parser.add_argument("--result-dir", default="user_data/backtest_results/auto-whitelist") parser.add_argument("--skip-download", action="store_true") parser.add_argument("--skip-backtest", action="store_true") parser.add_argument( "--candidates-only", action="store_true", help="Print the live candidate/liquidity snapshot without writing files or running Docker.", ) parser.add_argument("--apply", action="store_true", help="Write selected whitelist to --base-config.") parser.add_argument("--restart", action="store_true", help="Restart docker compose service after --apply.") parser.add_argument("--service", default="freqtrade") return parser.parse_args() def main() -> int: args = parse_args() project_dir = Path(args.project_dir).resolve() base_config = project_dir / args.base_config work_config = project_dir / args.work_config result_dir = project_dir / args.result_dir result_dir.mkdir(parents=True, exist_ok=True) start = args.start_date or utc_yyyymmdd(args.days) timerange = f"{start}-{args.end_date}" if args.end_date else f"{start}-" include = load_list(args.include) exclude = load_list(args.exclude) keep = load_list(args.keep) candidates = current_top_pairs( max_candidates=args.max_candidates, min_quote_volume=args.min_quote_volume, include=include, exclude=exclude, ) candidate_pairs = [row["pair"] for row in candidates] if not candidate_pairs: raise RuntimeError("No candidate pairs found.") print("Candidate pairs:") for index, row in enumerate(candidates, 1): volume = row["quote_volume"] forced = " forced" if row.get("forced") else "" liquidity = " BELOW_MIN_VOLUME" if row.get("below_min_quote_volume") else "" print( f"{index:02d}. {row['pair']:<22} {row['symbol']:<14}" f" quoteVolume={volume:,.0f}{forced}{liquidity}" ) if args.candidates_only: return 0 write_config( base_config=base_config, output_config=work_config, pairs=candidate_pairs, bot_name=f"auto-{args.strategy.lower()}-backtest", ) print(f"Wrote temporary config: {work_config}") config_arg = str(work_config.relative_to(project_dir)) if not args.skip_download: run( [ "docker", "compose", "run", "--rm", args.service, "download-data", "--config", config_arg, "--timerange", timerange, "--timeframes", *args.timeframes, ], cwd=project_dir, ) result: dict[str, Any] | None = None zip_path: Path | None = None if not args.skip_backtest: run( build_backtest_command( service=args.service, config_arg=config_arg, strategy=args.strategy, strategy_path=args.strategy_path, timerange=timerange, timeframe=args.timeframe, fee=args.fee, result_dir_arg=str(result_dir.relative_to(project_dir)), ), cwd=project_dir, ) zip_path = newest_backtest_zip(result_dir) result = load_backtest(zip_path, args.strategy) else: zip_path = newest_backtest_zip(result_dir) result = load_backtest(zip_path, args.strategy) selected = select_pairs( result, min_trades=args.select_min_trades, min_profit_pct=args.select_min_profit_pct, min_profit_factor=args.select_min_profit_factor, max_selected=args.select_max_pairs, keep=keep, ) selected_pairs = [row["pair"] for row in selected] report = { "created_at": dt.datetime.now(dt.timezone.utc).isoformat(), "timerange": timerange, "strategy": args.strategy, "timeframe": args.timeframe, "fee": args.fee, "candidates": candidates, "backtest_zip": str(zip_path) if zip_path else None, "summary": { "total_trades": result.get("total_trades"), "profit_total_pct": float(result.get("profit_total", 0) or 0) * 100, "profit_factor": result.get("profit_factor"), "max_drawdown_pct": float(result.get("max_drawdown_account", 0) or 0) * 100, "winrate": result.get("winrate"), }, "selection_rules": { "min_quote_volume": args.min_quote_volume, "min_trades": args.select_min_trades, "min_profit_pct": args.select_min_profit_pct, "min_profit_factor": args.select_min_profit_factor, "max_pairs": args.select_max_pairs, }, "selected": selected, } report_path = result_dir / "latest-selection-report.json" report_path.write_text(json.dumps(report, indent=2) + "\n") print("\nBacktest summary:") print(json.dumps(report["summary"], indent=2)) print("\nSelected whitelist:") for row in selected: print( f"{row['pair']:<22} trades={row['trades']:<4} " f"profit={row['profit_total_pct']:+.2f}% " f"pf={row['profit_factor']:.2f} " f"winrate={row['winrate'] * 100:.1f}%" ) print(f"\nReport: {report_path}") if args.apply: if not selected_pairs: raise RuntimeError("Refusing to apply an empty selected whitelist.") apply_whitelist(base_config, selected_pairs) if args.restart: run(["docker", "compose", "restart", args.service], cwd=project_dir) return 0 if __name__ == "__main__": try: raise SystemExit(main()) except KeyboardInterrupt: raise SystemExit(130) except Exception as exc: print(f"ERROR: {exc}", file=sys.stderr) raise SystemExit(1)