待翻译：Building and Validating a Quantitative Trading Strategy with OctoBot, Walk-Forward Backtesting, Parameter Optimization, and Interactive Analysis

AI 服务暂时不可用，以下为来源正文，待恢复后补全翻译。

In this tutorial, we build a complete quantitative backtesting workflow with OctoBot and OctoBot-Script while keeping the environment isolated from Colab’s preinstalled dependencies. We configure a rule-based trading strategy that combines RSI-based oversold signals, EMA trend confirmation, and ATR-driven adaptive stop-loss and take-profit levels, and we execute it through OctoBot’s native market-order and backtesting APIs. We also retrieve historical OHLCV data through OctoBot’s data layer with automatic exchange fallback, perform a multi-parameter grid search over an in-sample period, and select the strongest configuration based on its excess return relative to buy-and-hold. We then validate the selected parameters on a completely separate out-of-sample period to assess generalization and identify potential overfitting. Finally, we extract OctoBot’s backtest report data and use Pandas and Plotly to analyze parameter sensitivity, portfolio performance, price action, indicators, and execution results in an interactive Colab environment. Copy CodeCopiedUse a different Browser SYMBOL = "BTC/USDT" TIME_FRAME = "1d" EXCHANGES = ["binance", "kucoin", "okx", "bybit", "mexc", "kraken"] IN_SAMPLE = ("2019-01-01", "2023-01-01") OUT_OF_SAMPLE = ("2023-01-01", "2025-06-01") GRID = { "rsi_period": [7, 14, 21], "rsi_threshold": [25, 30, 35], "tp_atr_mult": [3.0, 5.0], } FIXED = { "ema_fast": 50, "ema_slow": 200, "atr_period": 14, "sl_atr_mult": 2.0, "position_size": "20%", "min_offset_pct": 1.0, "max_offset_pct": 40.0, } VENV_DIR = "/content/octobot_env" WORK_DIR = "/content/octobot_lab" OCTOBOT_V = "2.1.1" PY_VERSION = "3.12" import json, os, subprocess, sys, textwrap, time, itertools, shutil os.makedirs(WORK_DIR, exist_ok=True) PY = os.path.join(VENV_DIR, "bin", "python") MARKER = os.path.join(VENV_DIR, ".octobot_ready") def sh(cmd, kw): """Run a command, streaming its output live into the Colab cell.""" print(f"$ {' '.join(cmd)}") p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, kw) for line in p.stdout: print(" " + line.rstrip()) p.wait() if p.returncode != 0: raise RuntimeError(f"command failed ({p.returncode}): {' '.join(cmd)}") if not os.path.exists(MARKER): print("=" * 90, "\n BUILDING OCTOBOT ENVIRONMENT (one-off, ~2 min)\n", "=" * 90) subprocess.run([sys.executable, "-m", "pip", "install", "-q", "uv"], check=True) UV = [sys.executable, "-m", "uv"] sh(UV + ["venv", "--python", PY_VERSION, VENV_DIR]) sh(UV + ["pip", "install", "--python", PY, "-q", f"OctoBot=={OCTOBOT_V}", "wheel", "setuptools", "appdirs==1.4.4"]) sh(UV + ["pip", "install", "--python", PY, "-q", "--no-build-isolation", "octobot-script"]) sh([PY, "-m", "octobot_script.cli", "install_tentacles", "--quite"]) sh([PY, "-c", textwrap.dedent(""" import os, shutil, octobot_script.resources as r base = r.get_report_resource_path("") src, dst_dir = os.path.join(base, "index.html"), os.path.join(base, "dist") os.makedirs(dst_dir, exist_ok=True) dst = os.path.join(dst_dir, "index.html") if os.path.exists(src) and not os.path.exists(dst): shutil.copy2(src, dst); print("patched report template ->", dst) else: print("report template already fine") """)]) open(MARKER, "w").write("ok") print("\n environment ready\n") else: print(" environment already built (delete", VENV_DIR, "to rebuild)\n") We define the core trading configuration, including the symbol, timeframe, exchange fallback list, backtesting windows, parameter grid, and fixed strategy settings. We then create an isolated Python environment with uv and install the pinned OctoBot and OctoBot-Script dependencies required for the workflow. We also install the OctoBot tentacles package and patch the report-template path so later backtest reporting works correctly inside the Colab environment. Copy CodeCopiedUse a different Browser WORKER = os.path.join(WORK_DIR, "octobot_worker.py") WORKER_SRC = r''' import asyncio, itertools, json, os, sys, time, traceback import numpy as np import tulipy import octobot_script as obs CFG = json.load(open(os.environ["OBS_CONFIG"])) OUT = os.environ["OBS_OUT"] FIX = CFG["fixed"] for kw in ("Close", "High", "Low", "Time", "market", "current_live_time", "plot_indicator"): if not hasattr(obs, kw): raise RuntimeError( f"octobot_script.{kw} missing -> tentacles are not installed. " "Run: python -m octobot_script.cli install_tentacles" ) def tail(*arrays): """tulipy indicators return different lengths; right-align them all.""" n = min(len(a) for a in arrays) return [np.asarray(a)[-n:] for a in arrays] def clamp(v): return float(min(max(v, FIX["min_offset_pct"]), FIX["max_offset_pct"])) def build_callbacks(params, run_data): """ OctoBot-Script splits a strategy into: initialize(ctx) -> runs once on the first candle. Do vectorised work here. strategy(ctx) -> runs on EVERY closed candle. Keep it cheap. """ async def initialize(ctx): closes = await obs.Close(ctx, max_history=True) highs = await obs.High(ctx, max_history=True) lows = await obs.Low(ctx, max_history=True) times = await obs.Time(ctx, max_history=True, use_close_time=True) rsi = tulipy.rsi(closes, period=params["rsi_period"]) ema_f = tulipy.ema(closes, period=FIX["ema_fast"]) ema_s = tulipy.ema(closes, period=FIX["ema_slow"]) atr = tulipy.atr(highs, lows, closes, period=FIX["atr_period"]) t, c, rsi, ema_f, ema_s, atr = tail(times, closes, rsi, ema_f, ema_s, atr) atr_pct = np.where(c > 0, atr / c * 100.0, 0.0) entries, offsets = set(), {} for i in range(len(t)): oversold = rsi[i] ema_s[i] if oversold and uptrend and atr_pct[i] > 0: ts = float(t[i]) entries.add(ts) offsets[ts] = ( clamp(FIX["sl_atr_mult"] * atr_pct[i]), clamp(params["tp_atr_mult"] * atr_pct[i]), ) run_data["entries"] = entries run_data["offsets"] = offsets if run_data.get("plot"): await obs.plot_indicator(ctx, f"RSI({params['rsi_period']})", t, rsi, entries) await obs.plot_indicator(ctx, f"EMA{FIX['ema_fast']}", t, ema_f) await obs.plot_indicator(ctx, f"EMA{FIX['ema_slow']}", t, ema_s) await obs.plot_indicator(ctx, "ATR %", t, atr_pct) async def strategy(ctx): now = obs.current_live_time(ctx) if now not in run_data["entries"]: return sl, tp = run_data["offsets"][now] await obs.market( ctx, "buy", amount=FIX["position_size"], stop_loss_offset=f"-{sl:.2f}%", take_profit_offset=f"{tp:.2f}%", ) return initialize, strategy def metrics(res): br = res.report.get("bot_report", {}) first = lambda d: float(list(d.values())[0]) if isinstance(d, dict) and d else float("nan") return { "profitability": first(br.get("profitability", {})), "market": first(br.get("market_average_profitability", {})), "reference": br.get("reference_market"), "start_portfolio": str(br.get("starting_portfolio")), "end_portfolio": str(br.get("end_portfolio")), "candles": res.candles_count, "duration_s": round(res.duration or 0, 2), "errors": res.report.get("errors_count"), } async def load_data(window): """Try each exchange until one serves data (Binance blocks many datacenter IPs).""" start, end = window last = None for ex in CFG["exchanges"]: try: print(f" ↓ fetching {CFG['symbol']} {CFG['time_frame']} from {ex} " f"[{time.strftime('%Y-%m-%d', time.gmtime(start))} → " f"{time.strftime('%Y-%m-%d', time.gmtime(end))}]", flush=True) data = await obs.get_data( CFG["symbol"], CFG["time_frame"], exchange=ex, exchange_type="spot", start_timestamp=start, end_timestamp=end, social_services=[], ) print(f" ✓ {ex} ok -> {data.data_files}", flush=True) return data, ex except Exception as e: last = e print(f" ✗ {ex}: {type(e).name}: {e}", flush=True) raise RuntimeError(f"no exchange served data; last error: {last}") async def backtest(data, params, plot=False, storage=False): run_data = {"entries": None, "offsets": {}, "plot": plot} init_f, strat_f = build_callbacks(params, run_data) res = await obs.run( data, params, strategy_func=strat_f, initialize_func=init_f, enable_logs=False, enable_storage=storage, ) return res, len(run_data["entries"] or ()) async def main(): out = {"grid": [], "best": None, "oos": None, "errors": []} print("\n" + "=" * 78 + "\n IN-SAMPLE GRID SEARCH\n" + "=" * 78, flush=True) is_data, ex_used = await load_data(CFG["in_sample"]) out["exchange"] = ex_used keys = list(CFG["grid"].keys()) combos = [dict(zip(keys, v)) for v in itertools.product(*CFG["grid"].values())] print(f" {len(combos)} configurations to evaluate\n", flush=True) for i, params in enumerate(combos, 1): try: res, n_sig = await backtest(is_data, params) m = metrics(res) m.update(params); m["signals"] = n_sig m["edge"] = m["profitability"] - m["market"] out["grid"].append(m) print(f" [{i:>2}/{len(combos)}] {params} " f"P&L {m['profitability']:+.2f}% vs market {m['market']:+.2f}% " f"edge {m['edge']:+.2f}% ({n_sig} signals, {m['duration_s']}s)", flush=True) except Exception as e: out["errors"].append(f"{params}: {e}") print(f" [{i:>2}/{len(combos)}] {params} FAILED: {e}", flush=True) traceback.print_exc() await is_data.stop() if not out["grid"]: json.dump(out, open(OUT, "w")); raise SystemExit("no successful runs") best = max(out["grid"], key=lambda r: r["edge"]) out["best"] = {k: best[k] for k in keys} print(f"\n best in-sample config: {out['best']} (edge {best['edge']:+.2f}%)", flush=True) print("\n" + "=" * 78 + "\n OUT-OF-SAMPLE VALIDATION (never optimised on)\n" + "=" * 78, flush=True) oos_data, _ = await load_data(CFG["out_of_sample"]) res, n_sig = await backtest(oos_data, out["best"], plot=True, storage=True) m = metrics(res); m.update(out["best"]) m["signals"] = n_sig; m["edge"] = m["profitability"] - m["market"] out["oos"] = m print(f" OOS P&L {m['profitability']:+.2f}% vs market {m['market']:+.2f}% " f"edge {m['edge']:+.2f}% ({n_sig} signals)", flush=True) print(" " + res.describe(), flush=True) report_dir = os.path.join(os.getcwd(), "report") os.makedirs(report_dir, exist_ok=True) try: plot = await res.plot(report_file=os.path.join(report_dir, "report.html"), show=False) out["bundle"] = os.path.join(os.path.dirname(os.path.abspath(plot.report_file)), "report.json") print(f" ✓ report bundle: {out['bundle']}", flush=True) except Exception as e: out["errors"].append(f"report: {e}") print(f" ✗ report generation failed: {e}", flush=True) await oos_data.stop() json.dump(out, open(OUT, "w"), indent=2, default=str) print("\n✓ results written to", OUT, flush=True) asyncio.run(main()) ''' with open(WORKER, "w") as f: f.write(WORKER_SRC) We build the standalone OctoBot worker that contains the strategy logic and executes inside the isolated virtual environment. We calculate RSI, fast and slow EMAs, and ATR values, generate entry signals when oversold conditions align with an upward trend, and derive volatility-adjusted stop-loss and take-profit offsets. We also define the historical data loader, backtest runner, grid-search loop, out-of-sample validation, performance metrics, and report generation process. Copy CodeCopiedUse a different Browser import datetime as _dt def ts(d): return int(_dt.datetime.strptime(d, "%Y-%m-%d") .replace(tzinfo=_dt.timezone.utc).timestamp()) CONFIG_PATH = os.path.join(WORK_DIR, "config.json") RESULTS_PATH = os.path.join(WORK_DIR, "results.json") json.dump({ "symbol": SYMBOL, "time_frame": TIME_FRAME, "exchanges": EXCHANGES, "in_sample": [ts(IN_SAMPLE[0]), ts(IN_SAMPLE[1])], "out_of_sample": [ts(OUT_OF_SAMPLE[0]), ts(OUT_OF_SAMPLE[1])], "grid": GRID, "fixed": FIXED, }, open(CONFIG_PATH, "w"), indent=2) env = dict(os.environ, OBS_CONFIG=CONFIG_PATH, OBS_OUT=RESULTS_PATH, PYTHONUNBUFFERED="1") t0 = time.time() proc = subprocess.Popen([PY, WORKER], cwd=WORK_DIR, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=1) for line in proc.stdout: print(line.rstrip()) proc.wait() print(f"\n total backtesting time: {time.time() - t0:.1f}s (exit {proc.returncode})") if not os.path.exists(RESULTS_PATH): raise SystemExit("No results produced — read the log above. " "Most common cause: every exchange refused the data request.") R = json.load(open(R [truncated for AI cost control]