Xây dựng và xác thực chiến lược giao dịch định lượng bằng OctoBot, kiểm thử ngược theo phương pháp Walk-Forward, tối ưu hóa tham số và phân tích tương tác

Trong hướng dẫn này, chúng tôi xây dựng một quy trình kiểm thử ngược (backtesting) định lượng hoàn chỉnh bằng OctoBot và OctoBot-Script, đồng thời giữ môi trường tách biệt khỏi các phần phụ thuộc được cài đặt sẵn của Colab. Chúng tôi cấu hình một chiến lược giao dịch dựa trên quy tắc, kết hợp các tín hiệu quá bán dựa trên chỉ báo RSI, xác nhận xu hướng bằng EMA, và các mức cắt lỗ (stop-loss) và chốt lời (take-profit) thích ứng dựa trên ATR. Chiến lược này được thực thi thông qua các API lệnh thị trường (market-order) và kiểm thử ngược gốc của OctoBot. Chúng tôi cũng truy xuất dữ liệu OHLCV lịch sử thông qua lớp dữ liệu của OctoBot với cơ chế dự phòng sàn giao dịch tự động, thực hiện tìm kiếm lưới đa tham số (multi-parameter grid search) trong một khoảng thời gian mẫu (in-sample period), và chọn cấu hình mạnh nhất dựa trên lợi nhuận vượt trội so với chiến lược mua và giữ (buy-and-hold). Sau đó, chúng tôi xác thực các tham số đã chọn trên một khoảng thời gian ngoài mẫu (out-of-sample period) hoàn toàn riêng biệt để đánh giá khả năng tổng quát hóa và xác định khả năng quá khớp (overfitting) tiềm ẩn. Cuối cùng, chúng tôi trích xuất dữ liệu báo cáo kiểm thử ngược của OctoBot và sử dụng Pandas cùng Plotly để phân tích độ nhạy tham số, hiệu suất danh mục đầu tư, biến động giá, các chỉ báo và kết quả thực thi trong môi trường Colab tương tác. ``` 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") ``` Chúng tôi định nghĩa cấu hình giao dịch cốt lõi, bao gồm mã giao dịch, khung thời gian, danh sách dự phòng sàn giao dịch, các cửa sổ kiểm thử ngược (backtesting windows), lưới tham số và các cài đặt chiến lược cố định. Sau đó, chúng tôi tạo một môi trường Python biệt lập bằng `uv` và cài đặt các phần phụ thuộc `OctoBot` và `OctoBot-Script` đã được ghim phiên bản, cần thiết cho quy trình làm việc. Chúng tôi cũng cài đặt gói `OctoBot tentacles` và vá đường dẫn `report-template` để việc báo cáo kiểm thử ngược sau này hoạt động chính xác trong môi trường Colab. ```python 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] < params["rsi_threshold"] uptrend = ema_f[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, entrie ```