使用 OctoBot 建構與驗證量化交易策略：前進式回測、參數最佳化及互動式分析

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 BrowserSYMBOL = "BTC/USDT" TIMEFRAME = "1d" EXCHANGES = ["binance", "kucoin", "okx", "bybit", "mexc", "kraken"] INSAMPLE = ("2019-01-01", "2023-01-01") OUTOFSAMPLE = ("2023-01-01", "2025-06-01") GRID = { "rsiperiod": [7, 14, 21], "rsithreshold": [25, 30, 35], "tpatrmult": [3.

0, 5.0], } FIXED = { "emafast": 50, "emaslow": 200, "atrperiod": 14, "slatrmult": 2.0, "positionsize": "20%", "minoffsetpct": 1.0, "maxoffsetpct": 40.0, } VENVDIR = "/content/octobotenv" WORKDIR = "/content/octobotlab" OCTOBOTV = "2.1.1" PYVERSION = "3.

12" import json, os, subprocess, sys, textwrap, time, itertools, shutil os.makedirs(WORKDIR, existok=True) PY = os.path.join(VENVDIR, "bin", "python") MARKER = os.path.join(VENVDIR, ".octobotready") 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", PYVERSION, VENVDIR]) sh(UV + ["pip", "install", "--python", PY, "-q", f"OctoBot=={OCTOBOTV}", "wheel", "setuptools", "appdirs==1.4.

4"]) sh(UV + ["pip", "install", "--python", PY, "-q", "--no-build-isolation", "octobot-script"]) sh([PY, "-m", "octobotscript.cli", "installtentacles", "--quite"]) sh([PY, "-c", textwrap.dedent(""" import os, shutil, octobotscript.resources as r base = r.getreportresourcepath("") src, dstdir = os.

path.join(base, "index.html"), os.path.join(base, "dist") os.makedirs(dstdir, existok=True) dst = os.path.join(dstdir, "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", VENVDIR, "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 BrowserWORKER = os.path.join(WORKDIR, "octobotworker.py") WORKERSRC = r''' import asyncio, itertools, json, os, sys, time, traceback import numpy as np import tulipy import octobotscript as obs CFG = json.load(open(os.environ["OBSCONFIG"])) OUT = os.

environ["OBSOUT"] FIX = CFG["fixed"] for kw in ("Close", "High", "Low", "Time", "market", "currentlivetime", "plotindicator"): if not hasattr(obs, kw): raise RuntimeError( f"octobotscript.{kw} missing -> tentacles are not installed." "Run: python -m octobotscript.

cli installtentacles" ) 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["minoffsetpct"]), FIX["maxoffsetpct"])) def buildcallbacks(params, rundata): """ 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, maxhistory=True) highs = await obs.High(ctx, maxhistory=True) lows = await obs.Low(ctx, maxhistory=True) times = await obs.

Time(ctx, maxhistory=True, useclosetime=True) rsi = tulipy.rsi(closes, period=params["rsiperiod"]) emaf = tulipy.ema(closes, period=FIX["emafast"]) emas = tulipy.ema(closes, period=FIX["emaslow"]) atr = tulipy.

atr(highs, lows, closes, period=FIX["atrperiod"]) t, c, rsi, emaf, emas, atr = tail(times, closes, rsi, emaf, emas, atr) atrpct = np.where(c > 0, atr / c 100.0, 0.

0) entries, offsets = set(), {} for i in range(len(t)): oversold = rsi[i] < params["rsithreshold"] uptrend = emaf[i] > emas[i] if oversold and uptrend and atrpct[i] > 0: ts = float(t[i]) entries.

add(ts) offsets[ts] = ( clamp(FIX["slatrmult"] atrpct[i]), clamp(params["tpatrmult"] atrpct[i]), ) rundata["entries"] = entries rundata["offsets"] = offsets if rundata.get("plot"): await obs.plotindicator(ctx, f"RSI({params['rsiperiod']})", t, rsi, entries) await obs.

plotindicator(ctx, f"EMA{FIX['emafast']}", t, emaf) await obs.plotindicator(ctx, f"EMA{FIX['emaslow']}", t, emas) await obs.plotindicator(ctx, "ATR %", t, atrpct) async def strategy(ctx): now = obs.

currentlivetime(ctx) if now not in rundata["entries"]: return sl, tp = rundata["offsets"][now] await obs.market( ctx, "buy", amount=FIX["positionsize"], stoplossoffset=f"-{sl:.2f}%", takeprofitoffset=f"{tp:.2f}%", ) return initialize, strategy def metrics(res): br = res.report.

get("botreport", {}) 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("marketaverageprofitability", {})), "reference": br.get("referencemarket"), "startportfolio": str(br.

get("startingportfolio")), "endportfolio": str(br.get("endportfolio")), "candles": res.candlescount, "durations": round(res.duration or 0, 2), "errors": res.report.get("errorscount"), } async def loaddata(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['timeframe']} 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.

getdata( CFG["symbol"], CFG["timeframe"], exchange=ex, exchangetype="spot", starttimestamp=start, endtimestamp=end, socialservices=[], ) print(f" ✓ {ex} ok -> {data.datafiles}", 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): rundata = {"entries": None, "offsets": {}, "plot": plot} initf, stratf = buildcallbacks(params, rundata) res = await obs.

run( data, params, strategyfunc=stratf, initializefunc=initf, enablelogs=False, enablestorage=storage, ) return res, len(rundata["entries"] or (