"""MACD parameter sweep — one fresh handle per backtest."""
from __future__ import annotations
import csv, ctypes, sys, time
from itertools import product
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent / "../../tutorial"))
from run import BarC, ReportC
ROOT = Path(__file__).resolve().parent.parent.parent
SO = ROOT / "tutorial" / "macd" / "strategy.so"
OHLCV = ROOT / "tutorial" / "data" / "btcusdt_15m_7d.csv"
def load_bars():
with OHLCV.open(newline="") as f:
rows = list(csv.DictReader(f))
n = len(rows)
bars = (BarC * n)()
for i, r in enumerate(rows):
bars[i] = BarC(float(r["open"]), float(r["high"]), float(r["low"]),
float(r["close"]), float(r["volume"]), int(r["timestamp"]))
return bars, n
def make_lib():
lib = ctypes.CDLL(str(SO))
lib.strategy_create.argtypes = [ctypes.c_char_p]
lib.strategy_create.restype = ctypes.c_void_p
lib.strategy_set_input.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_char_p]
lib.strategy_set_override.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_char_p]
lib.run_backtest_full.argtypes = [
ctypes.c_void_p, ctypes.POINTER(BarC), ctypes.c_int,
ctypes.c_char_p, ctypes.c_char_p,
ctypes.c_int, ctypes.c_int, ctypes.c_int,
ctypes.POINTER(ReportC)]
lib.report_free.argtypes = [ctypes.POINTER(ReportC)]
lib.strategy_free.argtypes = [ctypes.c_void_p]
return lib
def run_one(lib, bars, n, inputs: dict, overrides: dict) -> ReportC:
"""One backtest with a freshly-allocated strategy handle."""
state = lib.strategy_create(b"{}")
for k, v in overrides.items():
lib.strategy_set_override(state, k.encode(), str(v).encode())
for k, v in inputs.items():
lib.strategy_set_input(state, k.encode(), str(v).encode())
report = ReportC()
lib.run_backtest_full(state, bars, n, b"", b"", 0, 4, 3,
ctypes.byref(report))
lib.strategy_free(state)
return report
def main() -> int:
if not SO.exists():
sys.exit("strategy.so missing — run `bash tutorial/run.sh` first")
lib = make_lib()
bars, n = load_bars()
overrides = {"initial_capital": 10000, "commission_value": 0.04}
fasts, slows, signals = [8, 12, 16], [21, 26, 32, 40], [9]
combos = [(f, sl, sg) for f, sl, sg in product(fasts, slows, signals) if f < sl]
print(f"sweeping {len(combos)} combos")
results, t0 = [], time.time()
for fast, slow, sig in combos:
inputs = {"Fast Length": fast, "Slow Length": slow, "Signal Length": sig}
report = run_one(lib, bars, n, inputs, overrides)
wins = sum(1 for i in range(report.trades_len) if report.trades[i].pnl > 0)
win_pct = (wins / report.trades_len * 100) if report.trades_len else 0.0
results.append((fast, slow, sig, report.trades_len, report.net_profit, win_pct))
print(f" fast={fast:02d} slow={slow:02d} sig={sig:02d} "
f"trades={report.trades_len:3d} pnl={report.net_profit:+9.2f} "
f"win={win_pct:.1f}%")
lib.report_free(ctypes.byref(report))
elapsed = time.time() - t0
print(f"\n{len(combos)} runs in {elapsed*1000:.1f} ms "
f"({elapsed*1000/len(combos):.1f} ms/run)")
print("\ntop 3 by net pnl:")
for f, sl, sg, _, pnl, _ in sorted(results, key=lambda r: -r[4])[:3]:
print(f" ({f:2d},{sl:2d},{sg:2d}) pnl={pnl:+.2f}")
return 0
if __name__ == "__main__":
sys.exit(main())
Each window starts cold — same equity, same TA seeds, no carry-over trades from the previous window.