WORKER = os.path.be part of(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 ("Shut", "Excessive", "Low", "Time", "market", "current_live_time", "plot_indicator"):
if not hasattr(obs, kw):
increase RuntimeError(
f"octobot_script.{kw} lacking -> tentacles aren't put in. "
"Run: python -m octobot_script.cli install_tentacles"
)
def tail(*arrays):
"""tulipy indicators return totally different lengths; right-align all of them."""
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 technique into:
initialize(ctx) -> runs as soon as on the primary candle. Do vectorised work right here.
technique(ctx) -> runs on EVERY closed candle. Maintain it low-cost.
"""
async def initialize(ctx):
closes = await obs.Shut(ctx, max_history=True)
highs = await obs.Excessive(ctx, max_history=True)
lows = await obs.Low(ctx, max_history=True)
occasions = await obs.Time(ctx, max_history=True, use_close_time=True)
rsi = tulipy.rsi(closes, interval=params["rsi_period"])
ema_f = tulipy.ema(closes, interval=FIX["ema_fast"])
ema_s = tulipy.ema(closes, interval=FIX["ema_slow"])
atr = tulipy.atr(highs, lows, closes, interval=FIX["atr_period"])
t, c, rsi, ema_f, ema_s, atr = tail(occasions, closes, rsi, ema_f, ema_s, atr)
atr_pct = np.the place(c > 0, atr / c * 100.0, 0.0)
entries, offsets = set(), {}
for i in vary(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 technique(ctx):
now = obs.current_live_time(ctx)
if not in run_data["entries"]:
return
sl, tp = run_data["offsets"]1786476726
await obs.market(
ctx, "purchase",
quantity=FIX["position_size"],
stop_loss_offset=f"-{sl:.2f}%",
take_profit_offset=f"{tp:.2f}%",
)
return initialize, technique
def metrics(res):
br = res.report.get("bot_report", {})
first = lambda d: float(listing(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": spherical(res.length or 0, 2),
"errors": res.report.get("errors_count"),
}
async def load_data(window):
"""Strive every trade till one serves information (Binance blocks many datacenter IPs)."""
begin, finish = window
final = None
for ex in CFG["exchanges"]:
strive:
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)
information = await obs.get_data(
CFG["symbol"], CFG["time_frame"],
trade=ex, exchange_type="spot",
start_timestamp=begin, end_timestamp=finish,
social_services=[],
)
print(f" ✓ {ex} okay -> {information.data_files}", flush=True)
return information, ex
besides Exception as e:
final = e
print(f" ✗ {ex}: {sort(e).__name__}: {e}", flush=True)
increase RuntimeError(f"no trade served information; final error: {final}")
async def backtest(information, 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(
information, params,
strategy_func=strat_f,
initialize_func=init_f,
enable_logs=False,
enable_storage=storage,
)
return res, len(run_data["entries"] or ())
async def important():
out = {"grid": [], "finest": 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 = listing(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):
strive:
res, n_sig = await backtest(is_data, params)
m = metrics(res)
m.replace(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} indicators, {m['duration_s']}s)", flush=True)
besides 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.cease()
if not out["grid"]:
json.dump(out, open(OUT, "w")); increase SystemExit("no profitable runs")
finest = max(out["grid"], key=lambda r: r["edge"])
out["best"] = {okay: finest[k] for okay in keys}
print(f"n 🏆 finest in-sample config: {out['best']} (edge {finest['edge']:+.2f}%)", flush=True)
print("n" + "=" * 78 + "n OUT-OF-SAMPLE VALIDATION (by no means 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.replace(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} indicators)", flush=True)
print(" " + res.describe(), flush=True)
report_dir = os.path.be part of(os.getcwd(), "report")
os.makedirs(report_dir, exist_ok=True)
strive:
plot = await res.plot(report_file=os.path.be part of(report_dir, "report.html"), present=False)
out["bundle"] = os.path.be part of(os.path.dirname(os.path.abspath(plot.report_file)),
"report.json")
print(f" ✓ report bundle: {out['bundle']}", flush=True)
besides Exception as e:
out["errors"].append(f"report: {e}")
print(f" ✗ report era failed: {e}", flush=True)
await oos_data.cease()
json.dump(out, open(OUT, "w"), indent=2, default=str)
print("n✓ outcomes written to", OUT, flush=True)
asyncio.run(important())
'''
with open(WORKER, "w") as f:
f.write(WORKER_SRC)
