From 641e342e0804526911ff2d78f25b9ca14b76eef4 Mon Sep 17 00:00:00 2001 From: luisleo526 Date: Fri, 10 Jul 2026 19:48:12 +0800 Subject: [PATCH] Add provider-to-engine backtest harness --- README.md | 33 ++ pyproject.toml | 3 + scripts/backtest.py | 9 + src/pineforge_data/__init__.py | 12 + src/pineforge_data/backtest.py | 540 +++++++++++++++++++++++++++++ src/pineforge_data/cli/__init__.py | 1 + src/pineforge_data/cli/backtest.py | 179 ++++++++++ tests/test_backtest.py | 145 ++++++++ 8 files changed, 922 insertions(+) create mode 100755 scripts/backtest.py create mode 100644 src/pineforge_data/backtest.py create mode 100644 src/pineforge_data/cli/__init__.py create mode 100644 src/pineforge_data/cli/backtest.py create mode 100644 tests/test_backtest.py diff --git a/README.md b/README.md index 4d38199..81222e7 100644 --- a/README.md +++ b/README.md @@ -72,6 +72,39 @@ the same `LiveTradeProvider` contract later. after an engine warmup. `start_sequence` is the last accepted sequence, so the adapter emits `start_sequence + 1` next. +## Direct backtest harness + +`pineforge-backtest` fetches confirmed OHLCV through a data provider, packs the +normalized bars into the PineForge C ABI, and calls a compiled strategy library +directly. It does not create an intermediate CSV. + +```bash +pineforge-backtest \ + --strategy /path/to/strategy.so \ + --exchange kraken \ + --symbol BTC/USD \ + --timeframe 15m \ + --start 2026-07-01T00:00:00Z \ + --end 2026-07-08T00:00:00Z \ + --output report.json \ + --pretty +``` + +The JSON report contains data provenance, processed-bar counts, every closed +trade, all/long/short trade statistics, equity statistics, security-feed +diagnostics, optional trace values, and the complete equity curve. Unix +millisecond timestamps can be used instead of ISO-8601 values. + +Use `--provider-config config.json` for CCXT constructor options and +`--strategy-params inputs.json` for Pine input overrides. The provider config +file may contain credentials, so keep it outside version control. + +Provider implementations are organized by their strongest supported runtime. +The current Python bucket contains CCXT and the harness; native low-latency +providers will live in the C++ bucket. Both buckets must emit the same +normalized records, but an individual provider does not need implementations +in both languages. + ## Development ```bash diff --git a/pyproject.toml b/pyproject.toml index 932e933..3859e66 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,6 +22,9 @@ classifiers = [ ] dependencies = [] +[project.scripts] +pineforge-backtest = "pineforge_data.cli.backtest:main" + [project.optional-dependencies] ccxt = ["ccxt>=4.5.64,<5"] dev = [ diff --git a/scripts/backtest.py b/scripts/backtest.py new file mode 100755 index 0000000..6562479 --- /dev/null +++ b/scripts/backtest.py @@ -0,0 +1,9 @@ +#!/usr/bin/env python3 +"""Repository wrapper for the installed ``pineforge-backtest`` command.""" + +from __future__ import annotations + +from pineforge_data.cli.backtest import main + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/pineforge_data/__init__.py b/src/pineforge_data/__init__.py index 14d70cc..563cf8e 100644 --- a/src/pineforge_data/__init__.py +++ b/src/pineforge_data/__init__.py @@ -1,5 +1,12 @@ """Provider-neutral market and macro data contracts for PineForge.""" +from .backtest import ( + BacktestOptions, + BacktestReport, + EngineBacktestError, + MagnifierDistribution, + PineForgeBacktestRunner, +) from .engine import EngineStreamSink, PfBar, PfTradeTick, pack_bars, pack_trade_ticks from .errors import EngineStreamError from .models import Bar, Instrument, MacroObservation, TradeTick @@ -16,6 +23,8 @@ from .requests import BarRequest, MacroRequest, TradeSubscription __all__ = [ + "BacktestOptions", + "BacktestReport", "Bar", "BarRequest", "CcxtCapabilityError", @@ -23,6 +32,7 @@ "CcxtDependencyError", "CcxtError", "CcxtProvider", + "EngineBacktestError", "EngineStreamError", "EngineStreamSink", "HistoricalBarProvider", @@ -31,8 +41,10 @@ "MacroDataProvider", "MacroObservation", "MacroRequest", + "MagnifierDistribution", "PfBar", "PfTradeTick", + "PineForgeBacktestRunner", "TradeSubscription", "TradeTick", "pack_bars", diff --git a/src/pineforge_data/backtest.py b/src/pineforge_data/backtest.py new file mode 100644 index 0000000..2fe82e7 --- /dev/null +++ b/src/pineforge_data/backtest.py @@ -0,0 +1,540 @@ +"""Run normalized PineForge data directly through a compiled strategy library.""" + +from __future__ import annotations + +import ctypes +import json +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from enum import IntEnum +from itertools import pairwise +from math import isfinite +from pathlib import Path +from typing import TypeAlias + +from .engine import PfBar, pack_bars +from .models import Bar, Instrument + +JsonScalar: TypeAlias = str | int | float | bool | None +JsonValue: TypeAlias = JsonScalar | list["JsonValue"] | dict[str, "JsonValue"] + +EXPECTED_PF_ABI = 2 + + +class EngineBacktestError(RuntimeError): + """A PineForge historical backtest could not be completed.""" + + +class MagnifierDistribution(IntEnum): + """PineForge bar-magnifier path sampling distributions.""" + + UNIFORM = 0 + COSINE = 1 + TRIANGLE = 2 + ENDPOINTS = 3 + FRONT_LOADED = 4 + BACK_LOADED = 5 + + +@dataclass(frozen=True, slots=True) +class BacktestOptions: + """Runtime options passed to ``run_backtest_full``.""" + + input_timeframe: str = "" + script_timeframe: str = "" + bar_magnifier: bool = False + magnifier_samples: int = 4 + magnifier_distribution: MagnifierDistribution = MagnifierDistribution.ENDPOINTS + trace_enabled: bool = False + chart_timezone: str | None = None + + def __post_init__(self) -> None: + if self.magnifier_samples <= 0: + raise ValueError("magnifier_samples must be positive") + if self.input_timeframe and not self.input_timeframe.strip(): + raise ValueError("input_timeframe must not be whitespace") + if self.script_timeframe and not self.script_timeframe.strip(): + raise ValueError("script_timeframe must not be whitespace") + if self.chart_timezone is not None and not self.chart_timezone.strip(): + raise ValueError("chart_timezone must not be empty") + + +@dataclass(frozen=True, slots=True) +class BacktestReport: + """A detached, JSON-safe copy of ``pf_report_t``.""" + + summary: Mapping[str, JsonValue] + metrics: Mapping[str, Mapping[str, JsonValue]] + trades: tuple[Mapping[str, JsonValue], ...] + security_diagnostics: tuple[Mapping[str, JsonValue], ...] + trace: tuple[Mapping[str, JsonValue], ...] + equity_curve: tuple[Mapping[str, JsonValue], ...] + + def to_dict(self) -> dict[str, JsonValue]: + """Return a mutable JSON-shaped representation.""" + + return { + "summary": dict(self.summary), + "metrics": {name: dict(values) for name, values in self.metrics.items()}, + "trades": [dict(trade) for trade in self.trades], + "security_diagnostics": [dict(item) for item in self.security_diagnostics], + "trace": [dict(item) for item in self.trace], + "equity_curve": [dict(point) for point in self.equity_curve], + } + + +class _PfTrade(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), + ] + + +class _PfTradeStats(ctypes.Structure): + _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 _PfEquityStats(ctypes.Structure): + _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), + ("sharpe_tv", ctypes.c_double), + ("sortino_tv", 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 _PfMetrics(ctypes.Structure): + _fields_ = [ + ("all", _PfTradeStats), + ("longs", _PfTradeStats), + ("shorts", _PfTradeStats), + ("equity", _PfEquityStats), + ] + + +class _PfEquityPoint(ctypes.Structure): + _fields_ = [ + ("time_ms", ctypes.c_int64), + ("equity", ctypes.c_double), + ("open_profit", ctypes.c_double), + ] + + +class _PfSecurityDiagnostic(ctypes.Structure): + _fields_ = [ + ("sec_id", ctypes.c_int), + ("feed_count", ctypes.c_int64), + ("complete_count", ctypes.c_int64), + ("partial_count", ctypes.c_int64), + ] + + +class _PfTraceEntry(ctypes.Structure): + _fields_ = [ + ("timestamp", ctypes.c_int64), + ("bar_index", ctypes.c_int32), + ("name_id", ctypes.c_int32), + ("value", ctypes.c_double), + ] + + +class _PfReport(ctypes.Structure): + _fields_ = [ + ("total_trades", ctypes.c_int), + ("trades", ctypes.POINTER(_PfTrade)), + ("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_complete_total", ctypes.c_int64), + ("security_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(_PfSecurityDiagnostic)), + ("security_diag_len", ctypes.c_int), + ("trace", ctypes.POINTER(_PfTraceEntry)), + ("trace_len", ctypes.c_int), + ("trace_names", ctypes.POINTER(ctypes.c_char_p)), + ("trace_names_len", ctypes.c_int), + ("metrics", _PfMetrics), + ("equity_curve", ctypes.POINTER(_PfEquityPoint)), + ("equity_curve_len", ctypes.c_int64), + ] + + +_TRADE_STATS_FIELDS = ( + "num_trades", + "num_wins", + "num_losses", + "num_even", + "percent_profitable", + "net_profit", + "net_profit_pct", + "gross_profit", + "gross_profit_pct", + "gross_loss", + "gross_loss_pct", + "profit_factor", + "avg_trade", + "avg_trade_pct", + "avg_win", + "avg_win_pct", + "avg_loss", + "avg_loss_pct", + "ratio_avg_win_avg_loss", + "largest_win", + "largest_win_pct", + "largest_loss", + "largest_loss_pct", + "commission_paid", + "expectancy", + "max_consecutive_wins", + "max_consecutive_losses", + "avg_bars_in_trade", + "avg_bars_in_wins", + "avg_bars_in_losses", +) +_EQUITY_STATS_FIELDS = ( + "max_equity_drawdown", + "max_equity_drawdown_pct", + "max_equity_runup", + "max_equity_runup_pct", + "buy_hold_return", + "buy_hold_return_pct", + "sharpe_tv", + "sortino_tv", + "sharpe_bar", + "sortino_bar", + "cagr", + "calmar", + "recovery_factor", + "time_in_market_pct", + "open_pl", +) + + +def _json_number(value: float) -> float | None: + return value if isfinite(value) else None + + +def _numeric_structure( + structure: ctypes.Structure, field_names: Sequence[str] +) -> dict[str, JsonValue]: + values: dict[str, JsonValue] = {} + for name in field_names: + value = getattr(structure, name) + if isinstance(value, float): + values[name] = _json_number(value) + elif isinstance(value, int): + values[name] = value + else: + raise EngineBacktestError(f"unsupported report field type for {name}") + return values + + +def _decode(value: bytes | None) -> str: + return value.decode("utf-8", "replace") if value else "" + + +class PineForgeBacktestRunner: + """Thin owner-safe wrapper around one compiled PineForge strategy library.""" + + def __init__(self, library: ctypes.CDLL) -> None: + self._library = library + self._check_abi() + self._configure_signatures() + + @classmethod + def load(cls, strategy_library: str | Path) -> PineForgeBacktestRunner: + """Load a compiled strategy shared library from disk.""" + + path = Path(strategy_library).expanduser().resolve() + if not path.is_file(): + raise FileNotFoundError(f"strategy library not found: {path}") + return cls(ctypes.CDLL(str(path))) + + def _check_abi(self) -> None: + try: + self._library.pf_abi_version.argtypes = [] + self._library.pf_abi_version.restype = ctypes.c_int + actual = int(self._library.pf_abi_version()) + except AttributeError as exc: + raise EngineBacktestError( + "strategy library predates pf_abi_version; rebuild it with the current engine" + ) from exc + if actual != EXPECTED_PF_ABI: + raise EngineBacktestError( + f"PineForge ABI mismatch: strategy reports {actual}, expected {EXPECTED_PF_ABI}" + ) + + def _configure_signatures(self) -> None: + library = self._library + library.strategy_create.argtypes = [ctypes.c_char_p] + library.strategy_create.restype = ctypes.c_void_p + library.strategy_free.argtypes = [ctypes.c_void_p] + library.strategy_free.restype = None + library.run_backtest_full.argtypes = [ + ctypes.c_void_p, + ctypes.POINTER(PfBar), + ctypes.c_int, + ctypes.c_char_p, + ctypes.c_char_p, + ctypes.c_int, + ctypes.c_int, + ctypes.c_int, + ctypes.POINTER(_PfReport), + ] + library.run_backtest_full.restype = None + library.report_free.argtypes = [ctypes.POINTER(_PfReport)] + library.report_free.restype = None + if hasattr(library, "strategy_get_last_error"): + library.strategy_get_last_error.argtypes = [ctypes.c_void_p] + library.strategy_get_last_error.restype = ctypes.c_char_p + if hasattr(library, "strategy_set_trace_enabled"): + library.strategy_set_trace_enabled.argtypes = [ctypes.c_void_p, ctypes.c_int] + library.strategy_set_trace_enabled.restype = None + if hasattr(library, "strategy_set_input"): + library.strategy_set_input.argtypes = [ + ctypes.c_void_p, + ctypes.c_char_p, + ctypes.c_char_p, + ] + library.strategy_set_input.restype = None + if hasattr(library, "strategy_set_chart_timezone"): + library.strategy_set_chart_timezone.argtypes = [ctypes.c_void_p, ctypes.c_char_p] + library.strategy_set_chart_timezone.restype = None + if hasattr(library, "strategy_set_syminfo_timezone"): + library.strategy_set_syminfo_timezone.argtypes = [ctypes.c_void_p, ctypes.c_char_p] + library.strategy_set_syminfo_timezone.restype = None + if hasattr(library, "strategy_set_syminfo_session"): + library.strategy_set_syminfo_session.argtypes = [ctypes.c_void_p, ctypes.c_char_p] + library.strategy_set_syminfo_session.restype = None + + def _apply_context( + self, state: int | ctypes.c_void_p, instrument: Instrument, options: BacktestOptions + ) -> None: + library = self._library + if options.trace_enabled and hasattr(library, "strategy_set_trace_enabled"): + library.strategy_set_trace_enabled(state, 1) + chart_timezone = options.chart_timezone or instrument.timezone + if chart_timezone and hasattr(library, "strategy_set_chart_timezone"): + library.strategy_set_chart_timezone(state, chart_timezone.encode()) + if instrument.timezone and hasattr(library, "strategy_set_syminfo_timezone"): + library.strategy_set_syminfo_timezone(state, instrument.timezone.encode()) + if instrument.session and hasattr(library, "strategy_set_syminfo_session"): + library.strategy_set_syminfo_session(state, instrument.session.encode()) + + def _last_error(self, state: int | ctypes.c_void_p) -> str: + if not hasattr(self._library, "strategy_get_last_error"): + return "" + return _decode(self._library.strategy_get_last_error(state)) + + def run( + self, + bars: Sequence[Bar], + *, + instrument: Instrument, + options: BacktestOptions | None = None, + strategy_params: Mapping[str, JsonValue] | None = None, + ) -> BacktestReport: + """Run confirmed normalized bars and return a detached report.""" + + if not bars: + raise ValueError("bars must not be empty") + if len(bars) > 2**31 - 1: + raise ValueError("bar count exceeds the PineForge C ABI limit") + if any(left.timestamp_ms >= right.timestamp_ms for left, right in pairwise(bars)): + raise ValueError("bar timestamps must be strictly increasing") + + runtime = options or BacktestOptions() + packed = pack_bars(bars, instrument=instrument) + params_json = json.dumps( + dict(strategy_params or {}), separators=(",", ":"), allow_nan=False + ).encode() + state = self._library.strategy_create(params_json) + if not state: + raise EngineBacktestError("strategy_create returned a null handle") + + native_report = _PfReport() + try: + self._apply_context(state, instrument, runtime) + if hasattr(self._library, "strategy_set_input"): + for name, value in (strategy_params or {}).items(): + self._library.strategy_set_input( + state, + name.encode(), + str(value).encode(), + ) + self._library.run_backtest_full( + state, + packed, + len(packed), + runtime.input_timeframe.encode(), + runtime.script_timeframe.encode(), + int(runtime.bar_magnifier), + runtime.magnifier_samples, + int(runtime.magnifier_distribution), + ctypes.byref(native_report), + ) + error = self._last_error(state) + if error: + raise EngineBacktestError(error) + return self._copy_report(native_report) + finally: + self._library.report_free(ctypes.byref(native_report)) + self._library.strategy_free(state) + + def _copy_report(self, report: _PfReport) -> BacktestReport: + summary: dict[str, JsonValue] = { + "total_trades": report.total_trades, + "net_profit": _json_number(report.net_profit), + "input_bars_processed": report.input_bars_processed, + "script_bars_processed": report.script_bars_processed, + "security_feeds_total": report.security_feeds_total, + "security_complete_total": report.security_complete_total, + "security_partial_total": report.security_partial_total, + "magnifier_sub_bars_total": report.magnifier_sub_bars_total, + "magnifier_sample_ticks_total": report.magnifier_sample_ticks_total, + "input_tf_seconds": report.input_tf_seconds, + "script_tf_seconds": report.script_tf_seconds, + "script_tf_ratio": report.script_tf_ratio, + "needs_aggregation": bool(report.needs_aggregation), + "bar_magnifier_enabled": bool(report.bar_magnifier_enabled), + } + metrics = { + "all": _numeric_structure(report.metrics.all, _TRADE_STATS_FIELDS), + "longs": _numeric_structure(report.metrics.longs, _TRADE_STATS_FIELDS), + "shorts": _numeric_structure(report.metrics.shorts, _TRADE_STATS_FIELDS), + "equity": _numeric_structure(report.metrics.equity, _EQUITY_STATS_FIELDS), + } + trades: list[Mapping[str, JsonValue]] = [] + for index in range(report.trades_len): + trade = report.trades[index] + trades.append( + { + "entry_time": trade.entry_time, + "exit_time": trade.exit_time, + "entry_price": _json_number(trade.entry_price), + "exit_price": _json_number(trade.exit_price), + "pnl": _json_number(trade.pnl), + "pnl_pct": _json_number(trade.pnl_pct), + "is_long": bool(trade.is_long), + "max_runup": _json_number(trade.max_runup), + "max_drawdown": _json_number(trade.max_drawdown), + "qty": _json_number(trade.qty), + "commission": _json_number(trade.commission), + "entry_bar_index": trade.entry_bar_index, + "exit_bar_index": trade.exit_bar_index, + } + ) + + diagnostics: list[Mapping[str, JsonValue]] = [] + for index in range(report.security_diag_len): + diagnostic = report.security_diag[index] + diagnostics.append( + { + "sec_id": diagnostic.sec_id, + "feed_count": diagnostic.feed_count, + "complete_count": diagnostic.complete_count, + "partial_count": diagnostic.partial_count, + } + ) + + trace_names = [ + _decode(report.trace_names[index]) for index in range(report.trace_names_len) + ] + trace: list[Mapping[str, JsonValue]] = [] + for index in range(report.trace_len): + entry = report.trace[index] + name = trace_names[entry.name_id] if 0 <= entry.name_id < len(trace_names) else None + trace.append( + { + "timestamp": entry.timestamp, + "bar_index": entry.bar_index, + "name_id": entry.name_id, + "name": name, + "value": _json_number(entry.value), + } + ) + + equity_curve: list[Mapping[str, JsonValue]] = [] + for index in range(report.equity_curve_len): + point = report.equity_curve[index] + equity_curve.append( + { + "time_ms": point.time_ms, + "equity": _json_number(point.equity), + "open_profit": _json_number(point.open_profit), + } + ) + return BacktestReport( + summary, + metrics, + tuple(trades), + tuple(diagnostics), + tuple(trace), + tuple(equity_curve), + ) + + +if ctypes.sizeof(_PfReport) != 944 or _PfReport.metrics.offset != 160: + raise RuntimeError("unsupported platform ABI layout for PineForge reports") diff --git a/src/pineforge_data/cli/__init__.py b/src/pineforge_data/cli/__init__.py new file mode 100644 index 0000000..7e050da --- /dev/null +++ b/src/pineforge_data/cli/__init__.py @@ -0,0 +1 @@ +"""Command-line harnesses shipped with pineforge-data.""" diff --git a/src/pineforge_data/cli/backtest.py b/src/pineforge_data/cli/backtest.py new file mode 100644 index 0000000..318e05c --- /dev/null +++ b/src/pineforge_data/cli/backtest.py @@ -0,0 +1,179 @@ +"""Fetch provider OHLCV and run a PineForge strategy without intermediate files.""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import re +import sys +from datetime import datetime +from pathlib import Path +from typing import cast + +from ..backtest import BacktestOptions, JsonValue, PineForgeBacktestRunner +from ..models import Instrument +from ..providers import CcxtProvider +from ..requests import BarRequest + + +def parse_timestamp(value: str) -> int: + """Parse Unix milliseconds or a timezone-aware ISO-8601 timestamp.""" + + try: + timestamp = int(value) + except ValueError: + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise argparse.ArgumentTypeError(f"invalid timestamp: {value}") from exc + if parsed.tzinfo is None: + raise argparse.ArgumentTypeError("ISO timestamps must include a timezone") from None + timestamp = int(parsed.timestamp() * 1_000) + if timestamp < 0: + raise argparse.ArgumentTypeError("timestamps must be non-negative") + return timestamp + + +def ccxt_timeframe_to_pine(timeframe: str) -> str: + """Translate common CCXT timeframe spelling into Pine timeframe spelling.""" + + match = re.fullmatch(r"([1-9][0-9]*)([smhdwM])", timeframe) + if match is None: + raise ValueError(f"cannot translate CCXT timeframe to PineForge: {timeframe}") + count = int(match.group(1)) + unit = match.group(2) + if unit == "s": + return f"{count}S" + if unit == "m": + return str(count) + if unit == "h": + return str(count * 60) + return f"{count}{unit.upper() if unit != 'M' else unit}" + + +def _load_json_object(path: Path | None) -> dict[str, JsonValue]: + if path is None: + return {} + value = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(value, dict) or not all(isinstance(key, str) for key in value): + raise ValueError(f"JSON file must contain an object: {path}") + return cast(dict[str, JsonValue], value) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="pineforge-backtest", + description="Feed provider OHLCV directly into a compiled PineForge strategy", + ) + parser.add_argument("--strategy", type=Path, required=True, help="compiled strategy library") + parser.add_argument("--provider", choices=("ccxt",), default="ccxt") + parser.add_argument("--exchange", required=True, help="CCXT exchange id, such as kraken") + parser.add_argument("--symbol", required=True, help="CCXT unified symbol, such as BTC/USD") + parser.add_argument("--timeframe", required=True, help="CCXT timeframe, such as 15m or 1h") + parser.add_argument("--start", type=parse_timestamp, required=True) + parser.add_argument("--end", type=parse_timestamp, required=True) + parser.add_argument("--limit", type=int) + parser.add_argument("--timezone", default="UTC", help="IANA chart and exchange timezone") + parser.add_argument("--session", default="24x7", help="PineForge syminfo session") + parser.add_argument( + "--engine-timeframe", + help="PineForge input timeframe; defaults to a translation of --timeframe", + ) + parser.add_argument("--script-timeframe", default="") + parser.add_argument("--provider-config", type=Path, help="CCXT constructor options JSON file") + parser.add_argument("--strategy-params", type=Path, help="strategy parameters JSON file") + parser.add_argument("--bar-magnifier", action="store_true") + parser.add_argument("--magnifier-samples", type=int, default=4) + parser.add_argument("--trace", action="store_true") + parser.add_argument("--output", type=Path, help="report path; defaults to stdout") + parser.add_argument("--pretty", action="store_true", help="pretty-print report JSON") + return parser + + +async def run_harness(args: argparse.Namespace) -> dict[str, JsonValue]: + """Execute the provider-to-engine pipeline for parsed CLI arguments.""" + + provider_config = _load_json_object(args.provider_config) + strategy_params = _load_json_object(args.strategy_params) + instrument = Instrument( + args.symbol, + venue=args.exchange, + timezone=args.timezone, + session=args.session, + ) + request = BarRequest( + instrument, + args.timeframe, + args.start, + args.end, + limit=args.limit, + ) + async with CcxtProvider(args.exchange, config=provider_config) as provider: + bars = await provider.fetch_bars(request) + provider_name = provider.name + if not bars: + raise RuntimeError("provider returned no confirmed bars for the requested interval") + + engine_timeframe = args.engine_timeframe or ccxt_timeframe_to_pine(args.timeframe) + options = BacktestOptions( + input_timeframe=engine_timeframe, + script_timeframe=args.script_timeframe or engine_timeframe, + bar_magnifier=args.bar_magnifier, + magnifier_samples=args.magnifier_samples, + trace_enabled=args.trace, + chart_timezone=args.timezone, + ) + runner = PineForgeBacktestRunner.load(args.strategy) + report = runner.run( + bars, + instrument=instrument, + options=options, + strategy_params=strategy_params, + ) + return { + "provider": { + "name": provider_name, + "exchange": args.exchange, + "symbol": args.symbol, + "source_timeframe": args.timeframe, + }, + "data": { + "requested_start_ms": args.start, + "requested_end_ms": args.end, + "first_bar_ms": bars[0].timestamp_ms, + "last_bar_ms": bars[-1].timestamp_ms, + "bars": len(bars), + }, + "strategy": { + "library": str(args.strategy.expanduser().resolve()), + "input_timeframe": engine_timeframe, + "script_timeframe": options.script_timeframe, + }, + "backtest": report.to_dict(), + } + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + try: + payload = asyncio.run(run_harness(args)) + rendered = json.dumps( + payload, + indent=2 if args.pretty else None, + sort_keys=args.pretty, + allow_nan=False, + ) + if args.output is None: + print(rendered) + else: + args.output.write_text(f"{rendered}\n", encoding="utf-8") + return 0 + except Exception as exc: + print(f"error: {exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_backtest.py b/tests/test_backtest.py new file mode 100644 index 0000000..06bbfc0 --- /dev/null +++ b/tests/test_backtest.py @@ -0,0 +1,145 @@ +from __future__ import annotations + +import ctypes +import json +from math import nan +from typing import cast + +import pytest + +from pineforge_data import ( + BacktestOptions, + Bar, + EngineBacktestError, + Instrument, + PineForgeBacktestRunner, +) +from pineforge_data.backtest import _PfEquityPoint, _PfReport, _PfTrade +from pineforge_data.cli.backtest import ccxt_timeframe_to_pine, parse_timestamp + + +class FakeFunction: + def __init__(self, result: object = None, callback: object = None) -> None: + self.result = result + self.callback = callback + self.calls: list[tuple[object, ...]] = [] + self.argtypes: object = None + self.restype: object = None + + def __call__(self, *args: object) -> object: + self.calls.append(args) + if callable(self.callback): + return self.callback(*args) + return self.result + + +class FakeBacktestLibrary: + def __init__(self, *, abi: int = 2, error: bytes = b"") -> None: + self.pf_abi_version = FakeFunction(abi) + self.strategy_create = FakeFunction(123) + self.strategy_free = FakeFunction() + self.report_free = FakeFunction() + self.strategy_get_last_error = FakeFunction(error) + self.strategy_set_input = FakeFunction() + self.strategy_set_trace_enabled = FakeFunction() + self.strategy_set_chart_timezone = FakeFunction() + self.strategy_set_syminfo_timezone = FakeFunction() + self.strategy_set_syminfo_session = FakeFunction() + self.run_backtest_full = FakeFunction(callback=self._fill_report) + self._trades = (_PfTrade * 1)( + _PfTrade(1_000, 2_000, 10.0, 12.0, 20.0, 20.0, 1, 2.5, 0.5, 10.0, 0.0, 0, 1) + ) + self._equity = (_PfEquityPoint * 2)( + _PfEquityPoint(1_000, 10_000.0, 0.0), + _PfEquityPoint(2_000, 10_020.0, 0.0), + ) + + def _fill_report(self, *args: object) -> None: + native = ctypes.cast(args[-1], ctypes.POINTER(_PfReport)).contents + native.total_trades = 1 + native.trades = self._trades + native.trades_len = 1 + native.net_profit = 20.0 + native.input_bars_processed = 2 + native.script_bars_processed = 2 + native.input_tf_seconds = 60 + native.script_tf_seconds = 60 + native.script_tf_ratio = 1 + native.metrics.all.num_trades = 1 + native.metrics.all.num_wins = 1 + native.metrics.all.net_profit = 20.0 + native.metrics.all.profit_factor = nan + native.metrics.equity.open_pl = 0.0 + native.equity_curve = self._equity + native.equity_curve_len = 2 + + +def bars() -> list[Bar]: + instrument = Instrument("BTC/USD", venue="fake") + return [ + Bar(instrument, 1_000, 10.0, 12.0, 9.0, 11.0, 5.0, "fake"), + Bar(instrument, 2_000, 11.0, 13.0, 10.0, 12.0, 6.0, "fake"), + ] + + +def test_runner_returns_detached_json_safe_report() -> None: + fake = FakeBacktestLibrary() + runner = PineForgeBacktestRunner(cast(ctypes.CDLL, fake)) + instrument = bars()[0].instrument + + report = runner.run( + bars(), + instrument=instrument, + options=BacktestOptions( + input_timeframe="1", + script_timeframe="1", + trace_enabled=True, + chart_timezone="UTC", + ), + strategy_params={"length": 14}, + ) + payload = report.to_dict() + + assert payload["summary"]["net_profit"] == 20.0 # type: ignore[index] + assert payload["metrics"]["all"]["profit_factor"] is None # type: ignore[index] + assert payload["trades"][0]["is_long"] is True # type: ignore[index] + assert len(payload["equity_curve"]) == 2 # type: ignore[arg-type] + json.dumps(payload, allow_nan=False) + assert len(fake.report_free.calls) == 1 + assert len(fake.strategy_free.calls) == 1 + assert fake.strategy_set_trace_enabled.calls == [(123, 1)] + assert fake.strategy_set_input.calls == [(123, b"length", b"14")] + assert fake.strategy_set_syminfo_session.calls == [(123, b"24x7")] + + +def test_runner_surfaces_engine_error_and_releases_owners() -> None: + fake = FakeBacktestLibrary(error=b"strategy failed") + runner = PineForgeBacktestRunner(cast(ctypes.CDLL, fake)) + + with pytest.raises(EngineBacktestError, match="strategy failed"): + runner.run(bars(), instrument=bars()[0].instrument) + + assert len(fake.report_free.calls) == 1 + assert len(fake.strategy_free.calls) == 1 + + +def test_runner_rejects_abi_mismatch_and_unsorted_bars() -> None: + with pytest.raises(EngineBacktestError, match="ABI mismatch"): + PineForgeBacktestRunner(cast(ctypes.CDLL, FakeBacktestLibrary(abi=99))) + + runner = PineForgeBacktestRunner(cast(ctypes.CDLL, FakeBacktestLibrary())) + with pytest.raises(ValueError, match="strictly increasing"): + runner.run(list(reversed(bars())), instrument=bars()[0].instrument) + + +@pytest.mark.parametrize( + ("ccxt", "pine"), + [("15m", "15"), ("2h", "120"), ("1d", "1D"), ("1w", "1W"), ("1M", "1M")], +) +def test_ccxt_timeframe_conversion(ccxt: str, pine: str) -> None: + assert ccxt_timeframe_to_pine(ccxt) == pine + + +def test_timestamp_parser_accepts_unix_ms_and_iso_8601() -> None: + assert parse_timestamp("1000") == 1_000 + assert parse_timestamp("1970-01-01T00:00:01Z") == 1_000