run.py

#!/usr/bin/env python3
"""Tutorial backtest: load tutorial/macd/strategy.so, run it against
tutorial/data/btcusdt_15m_7d.csv, print summary stats."""
from __future__ import annotations

import csv
import ctypes
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

ROOT  = Path(__file__).resolve().parent
SO    = ROOT / "macd" / "strategy.so"
OHLCV = ROOT / "data" / "btcusdt_15m_7d.csv"


# ctypes mirror of <pineforge/pineforge.h>
class BarC(ctypes.Structure):
    _fields_ = [("open",  ctypes.c_double), ("high",   ctypes.c_double),
                ("low",   ctypes.c_double), ("close",  ctypes.c_double),
                ("volume",ctypes.c_double), ("timestamp", ctypes.c_int64)]

class TradeC(ctypes.Structure):
    _fields_ = [("entry_time",  ctypes.c_int64), ("exit_time",  ctypes.c_int64),
                ("entry_price", ctypes.c_double),("exit_price", ctypes.c_double),
                ("pnl",         ctypes.c_double),("pnl_pct",    ctypes.c_double),
                ("is_long",     ctypes.c_int),   ("max_runup",  ctypes.c_double),
                ("max_drawdown",ctypes.c_double),("qty",        ctypes.c_double),
                ("commission",  ctypes.c_double),
                ("entry_bar_index", ctypes.c_int32),
                ("exit_bar_index",  ctypes.c_int32),
                ("open_at_end",     ctypes.c_int32)]   # ABI v3: range-end close row

class TradeStatsC(ctypes.Structure):  # pf_trade_stats_t
    _fields_ = [("num_trades", ctypes.c_int32), ("num_wins", ctypes.c_int32),
                ("num_losses", ctypes.c_int32), ("num_even", ctypes.c_int32),
                ("percent_profitable", ctypes.c_double),
                ("net_profit", ctypes.c_double), ("net_profit_pct", ctypes.c_double),
                ("gross_profit", ctypes.c_double), ("gross_profit_pct", ctypes.c_double),
                ("gross_loss", ctypes.c_double), ("gross_loss_pct", ctypes.c_double),
                ("profit_factor", ctypes.c_double),
                ("avg_trade", ctypes.c_double), ("avg_trade_pct", ctypes.c_double),
                ("avg_win", ctypes.c_double), ("avg_win_pct", ctypes.c_double),
                ("avg_loss", ctypes.c_double), ("avg_loss_pct", ctypes.c_double),
                ("ratio_avg_win_avg_loss", ctypes.c_double),
                ("largest_win", ctypes.c_double), ("largest_win_pct", ctypes.c_double),
                ("largest_loss", ctypes.c_double), ("largest_loss_pct", ctypes.c_double),
                ("commission_paid", ctypes.c_double),
                ("expectancy", ctypes.c_double),
                ("max_consecutive_wins", ctypes.c_int32),
                ("max_consecutive_losses", ctypes.c_int32),
                ("avg_bars_in_trade", ctypes.c_double),
                ("avg_bars_in_wins", ctypes.c_double),
                ("avg_bars_in_losses", ctypes.c_double)]

class EquityStatsC(ctypes.Structure):  # pf_equity_stats_t
    _fields_ = [("max_equity_drawdown", ctypes.c_double),
                ("max_equity_drawdown_pct", ctypes.c_double),
                ("max_equity_runup", ctypes.c_double),
                ("max_equity_runup_pct", ctypes.c_double),
                ("buy_hold_return", ctypes.c_double),
                ("buy_hold_return_pct", ctypes.c_double),
                # The C field names. The JSON report keys of these two stay
                # sharpe_tv / sortino_tv (ADR-0001, "Deprecated public spellings").
                ("sharpe_monthly", ctypes.c_double), ("sortino_monthly", ctypes.c_double),
                ("sharpe_bar", ctypes.c_double), ("sortino_bar", ctypes.c_double),
                ("cagr", ctypes.c_double), ("calmar", ctypes.c_double),
                ("recovery_factor", ctypes.c_double),
                ("time_in_market_pct", ctypes.c_double),
                ("open_pl", ctypes.c_double)]

class MetricsC(ctypes.Structure):  # pf_metrics_t
    _fields_ = [("all", TradeStatsC), ("longs", TradeStatsC),
                ("shorts", TradeStatsC), ("equity", EquityStatsC)]

class EquityPointC(ctypes.Structure):  # pf_equity_point_t
    _fields_ = [("time_ms", ctypes.c_int64), ("equity", ctypes.c_double),
                ("open_profit", ctypes.c_double)]

class _Diag(ctypes.Structure):
    _fields_ = [("sec_id", ctypes.c_int), ("feed_count", ctypes.c_int64),
                ("eval_complete_count", ctypes.c_int64),
                ("eval_partial_count",  ctypes.c_int64)]

class _Trace(ctypes.Structure):
    _fields_ = [("timestamp", ctypes.c_int64), ("bar_index", ctypes.c_int32),
                ("name_id",   ctypes.c_int32), ("value",     ctypes.c_double)]

class ReportC(ctypes.Structure):
    _fields_ = [("total_trades", ctypes.c_int),
                ("trades", ctypes.POINTER(TradeC)), ("trades_len", ctypes.c_int),
                ("net_profit", ctypes.c_double),
                ("input_bars_processed",         ctypes.c_int64),
                ("script_bars_processed",        ctypes.c_int64),
                ("security_feeds_total",         ctypes.c_int64),
                ("security_eval_complete_total", ctypes.c_int64),
                ("security_eval_partial_total",  ctypes.c_int64),
                ("magnifier_sub_bars_total",     ctypes.c_int64),
                ("magnifier_sample_ticks_total", ctypes.c_int64),
                ("input_tf_seconds",  ctypes.c_int),
                ("script_tf_seconds", ctypes.c_int),
                ("script_tf_ratio",   ctypes.c_int),
                ("needs_aggregation", ctypes.c_int),
                ("bar_magnifier_enabled", ctypes.c_int),
                ("security_diag", ctypes.POINTER(_Diag)),
                ("security_diag_len", ctypes.c_int),
                ("trace", ctypes.POINTER(_Trace)), ("trace_len", ctypes.c_int),
                ("trace_names", ctypes.POINTER(ctypes.c_char_p)),
                ("trace_names_len", ctypes.c_int),
                ("metrics", MetricsC),
                ("equity_curve", ctypes.POINTER(EquityPointC)),
                ("equity_curve_len", ctypes.c_int64),  # int64, NOT c_int
                ("broker_state_hash", ctypes.POINTER(ctypes.c_uint64)),
                ("broker_state_hash_len", ctypes.c_int64)]


# pf_report_t is caller-allocated, so a stale mirror means the runtime
# writes past our buffer. Assert the .so's ABI version before any run.
# v4 appended the live-runtime accessors and grew pf_report_t with the
# broker_state_hash array after equity_curve_len (ReportC above already
# carries both fields).
EXPECTED_PF_ABI = 4

def check_abi(lib: ctypes.CDLL) -> None:
    try:
        lib.pf_abi_version.restype = ctypes.c_int
        abi = lib.pf_abi_version()
    except AttributeError:
        raise RuntimeError(
            "strategy .so predates pf_abi_version (ABI v1); rebuild it against "
            "the current pineforge runtime (pf_report_t grew).")
    if abi != EXPECTED_PF_ABI:
        raise RuntimeError(
            f"pineforge ABI mismatch: .so reports {abi}, harness expects "
            f"{EXPECTED_PF_ABI}; rebuild.")


# A failure's stable code and arguments (engine 1.4.0+; an older .so has neither),
# and whether the last run completed (strategy_last_run_status, ABI v4).
def declare_error_code(lib: ctypes.CDLL) -> None:
    for name in ("strategy_get_last_error_code", "strategy_get_last_error_args"):
        if hasattr(lib, name):
            getattr(lib, name).argtypes = [ctypes.c_void_p]
            getattr(lib, name).restype = ctypes.c_char_p
    if hasattr(lib, "strategy_last_run_status"):
        lib.strategy_last_run_status.argtypes = [ctypes.c_void_p]
        lib.strategy_last_run_status.restype = ctypes.c_int


def error_code(lib: ctypes.CDLL, state) -> str:
    """'<code> <args JSON>' of the last failure on state, or '' without one."""
    if not hasattr(lib, "strategy_get_last_error_code"):
        return ""
    code = lib.strategy_get_last_error_code(state)
    if not code:
        return ""
    args = (lib.strategy_get_last_error_args(state)
            if hasattr(lib, "strategy_get_last_error_args") else None)
    return f"{code.decode()} {args.decode() if args else '{}'}"


def run_error(lib: ctypes.CDLL, state) -> str | None:
    """'<text> (<code> <args JSON>)' when the run just made on state failed, else
    None. A run failed when the engine reports a text, a code or a run status of 1
    (strategy_last_run_status), as the release harness decides, so a script stopped
    by runtime.error("") fails instead of printing a result."""
    text = ""
    if hasattr(lib, "strategy_get_last_error"):
        raw = lib.strategy_get_last_error(state)
        text = raw.decode("utf-8", "replace") if raw else ""
    code = error_code(lib, state)
    status = (lib.strategy_last_run_status(state)
              if hasattr(lib, "strategy_last_run_status") else 0)
    if not (text or code or status == 1):
        return None
    if not (text or code):
        text = "the run did not complete and the engine reported no error"
    return " ".join(part for part in (text, f"({code})" if code else "") if part)


def main() -> int:
    if not SO.exists():
        sys.exit(f"strategy.so missing — run `bash tutorial/run.sh` first")

    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"]))

    lib = ctypes.CDLL(str(SO))
    check_abi(lib)
    lib.strategy_create.argtypes  = [ctypes.c_char_p]
    lib.strategy_create.restype   = ctypes.c_void_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.strategy_free.argtypes    = [ctypes.c_void_p]
    lib.report_free.argtypes      = [ctypes.POINTER(ReportC)]
    if hasattr(lib, "strategy_get_last_error"):
        lib.strategy_get_last_error.argtypes = [ctypes.c_void_p]
        lib.strategy_get_last_error.restype  = ctypes.c_char_p
    declare_error_code(lib)

    state, report = lib.strategy_create(b"{}"), ReportC()
    t0 = time.time()
    lib.run_backtest_full(state, bars, n, b"", b"", 0, 4, 3, ctypes.byref(report))
    elapsed = time.time() - t0
    failure = run_error(lib, state)
    if failure is not None:
        lib.report_free(ctypes.byref(report))
        lib.strategy_free(state)
        print(f"engine error: {failure}", file=sys.stderr)
        return 1

    pnls = [report.trades[i].pnl for i in range(report.trades_len)]
    wins, losses = sum(p > 0 for p in pnls), sum(p < 0 for p in pnls)
    cum = peak = max_dd = 0.0
    for p in pnls:
        cum += p; peak = max(peak, cum); max_dd = min(max_dd, cum - peak)

    fmt = lambda ms: datetime.fromtimestamp(ms / 1000, tz=timezone.utc).strftime("%Y-%m-%d %H:%M")
    print(f"MACD(12,26,9) on BTCUSDT 15m — {n} bars, "
          f"{fmt(bars[0].timestamp)} → {fmt(bars[-1].timestamp)} UTC")
    print(f"  trades:    {report.trades_len}  "
          f"({wins}W / {losses}L, {wins/report.trades_len*100 if report.trades_len else 0:.1f}% win)")
    print(f"  net pnl:   {report.net_profit:+.2f}")
    print(f"  best/worst:{(max(pnls) if pnls else 0):+.2f} / {(min(pnls) if pnls else 0):+.2f}")
    print(f"  max dd:    {max_dd:.2f}")
    print(f"  elapsed:   {elapsed*1000:.1f} ms")

    lib.report_free(ctypes.byref(report))
    lib.strategy_free(state)
    return 0


if __name__ == "__main__":
    sys.exit(main())