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

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, 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"]1786470893
       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 SEARCHn" + "=" * 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 evaluaten", 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)