mirror of
https://github.com/KhizarImran/backtestingfx.git
synced 2026-11-02 11:36:31 +00:00
KhizarImran
7627094c63
Adds two Broker methods for path-dependent strategies: - close_partial(id, lot_size) scales out of a position and leaves the remainder open under the same id, so it can keep being managed. - update_sl(id, stop_loss) moves the stop in place. A trailing stop no longer has to close and reopen, which paid spread and commission on every step. Both are reachable from Python as Strategy.close_partial and Strategy.update_sl. Closing a position, part of one, or the whole book previously repeated the same PnL and commission arithmetic in four places. That is now one settle() helper taking the number of lots to close, so a partial close is the same code with a smaller number and the feature made broker.rs shorter rather than longer. Commission splits across the legs: the entry was charged on the full size when the position opened, so settle only takes the exit leg for the lots being closed. Two 0.5 closes land on the same cash as one 1.0 close, which is what closing_partially_twice_matches_closing_once pins. update_sl deliberately does not check which way the stop moves, since widening a stop is a legitimate thing for a strategy to do.
181 lines
6.6 KiB
Python
181 lines
6.6 KiB
Python
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
|
|
import pandas as pd
|
|
|
|
from backtestingfx import Backtest, Strategy
|
|
from backtestingfx.backtest import _DataView
|
|
|
|
|
|
class DataViewTest(unittest.TestCase):
|
|
def test_window_hides_future_bars_and_slices_correctly(self):
|
|
view = _DataView(["a", "b", "c", "d"])
|
|
view._len = 3 # only the first 3 bars are "visible" so far
|
|
|
|
self.assertEqual(len(view), 3)
|
|
self.assertEqual(view[-1], "c") # last visible, not "d"
|
|
self.assertEqual(view[0], "a")
|
|
self.assertEqual(view[-2:], ["b", "c"]) # trailing slice, no "d"
|
|
self.assertEqual(list(view), ["a", "b", "c"])
|
|
with self.assertRaises(IndexError):
|
|
view[3] # "d" is in the future, not addressable yet
|
|
|
|
|
|
class BuyAndHold(Strategy):
|
|
def next(self):
|
|
if not self.positions:
|
|
self.buy(1.0)
|
|
|
|
|
|
class BacktestTest(unittest.TestCase):
|
|
def test_run_returns_stats_from_python_strategy(self):
|
|
data = pd.DataFrame(
|
|
{
|
|
"open": [1.1, 1.1],
|
|
"high": [1.1, 1.1],
|
|
"low": [1.1, 1.1],
|
|
"close": [1.1, 1.1],
|
|
},
|
|
index=pd.to_datetime(["2026-01-01 00:00", "2026-01-01 01:00"], utc=True),
|
|
)
|
|
|
|
backtest = Backtest(
|
|
data,
|
|
BuyAndHold,
|
|
cash=10_000.0,
|
|
commission=7.0,
|
|
spread=0.0,
|
|
)
|
|
stats = backtest.run()
|
|
|
|
self.assertEqual(stats.initial_cash, 10_000.0)
|
|
self.assertEqual(stats.final_cash, 9_986.0)
|
|
self.assertEqual(stats.num_trades, 1)
|
|
self.assertEqual(stats.avg_pnl, -14.0)
|
|
self.assertEqual(stats.equity_curve, [10_000.0, 9_993.0, 9_986.0])
|
|
self.assertEqual(len(stats.trades), 1)
|
|
self.assertEqual(stats.trades[0].pnl, -14.0)
|
|
self.assertEqual(
|
|
stats.trades[0].exit_timestamp - stats.trades[0].entry_timestamp,
|
|
3_600,
|
|
)
|
|
|
|
data.drop(index=data.index[-1], inplace=True)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
report = Path(
|
|
backtest.plot(Path(directory) / "report.html", open_browser=False)
|
|
)
|
|
contents = report.read_text(encoding="utf-8")
|
|
|
|
self.assertTrue(report.is_file())
|
|
self.assertIn("BuyAndHold | backtestingfx report", contents)
|
|
self.assertIn("Market replay", contents)
|
|
self.assertIn("Plotly.newPlot", contents)
|
|
self.assertIn("2026-01-01 00:00", contents)
|
|
self.assertIn("2026-01-01 01:00", contents)
|
|
|
|
|
|
class TrailAndScaleOut(Strategy):
|
|
"""Buy once, trail the stop up each bar, and scale half out on the third bar."""
|
|
|
|
def next(self):
|
|
if not self.positions:
|
|
self.buy(1.0, stop_loss=self._bar.close - 0.0100)
|
|
return
|
|
position = self.positions[0]
|
|
self.update_sl(position.id, self._bar.close - 0.0100)
|
|
if self.index == 3:
|
|
self.close_partial(position.id, 0.5)
|
|
|
|
|
|
class PartialAndTrailingStopTest(unittest.TestCase):
|
|
def test_scaling_out_and_trailing_the_stop_through_the_python_api(self):
|
|
backtest = Backtest(rising_market(bars=6), TrailAndScaleOut, cash=10_000.0)
|
|
stats = backtest.run()
|
|
|
|
# two trades from one position: the 0.5 scaled out, then the 0.5 liquidated
|
|
self.assertEqual(stats.num_trades, 2)
|
|
self.assertEqual([trade.lot_size for trade in stats.trades], [0.5, 0.5])
|
|
self.assertEqual(stats.trades[0].entry_price, stats.trades[1].entry_price)
|
|
|
|
def test_update_sl_reports_a_missing_position(self):
|
|
recorded = []
|
|
|
|
class Probe(Strategy):
|
|
def next(self):
|
|
recorded.append(self.update_sl(999, 1.0))
|
|
|
|
Backtest(rising_market(bars=2), Probe, cash=10_000.0).run()
|
|
self.assertEqual(recorded, [False, False])
|
|
|
|
|
|
def rising_market(bars=20):
|
|
closes = [1.1000 + 0.0010 * i for i in range(bars)]
|
|
return pd.DataFrame(
|
|
{"open": closes, "high": closes, "low": closes, "close": closes},
|
|
index=pd.date_range("2026-01-01", periods=bars, freq="h", tz="UTC"),
|
|
)
|
|
|
|
|
|
class OptimizeTest(unittest.TestCase):
|
|
def test_grid_runs_every_combo_and_ranks_by_metric(self):
|
|
def hold_lots(df, lots):
|
|
return [lots] * len(df)
|
|
|
|
backtest = Backtest(rising_market(), cash=10_000.0)
|
|
results = backtest.optimize(hold_lots, lots=[0.1, 0.5, 1.0])
|
|
|
|
self.assertEqual(len(results), 3)
|
|
# price only rises, so the biggest long wins and ranking is strictly descending
|
|
self.assertEqual([params["lots"] for params, _ in results], [1.0, 0.5, 0.1])
|
|
returns = [stats.total_return_pct for _, stats in results]
|
|
self.assertEqual(returns, sorted(returns, reverse=True))
|
|
|
|
def test_grid_is_the_cartesian_product_and_matches_a_single_run(self):
|
|
def hold_lots(df, lots, unused):
|
|
return [lots] * len(df)
|
|
|
|
backtest = Backtest(rising_market(), cash=10_000.0)
|
|
results = backtest.optimize(hold_lots, lots=[0.1, 0.2], unused=["a", "b"])
|
|
self.assertEqual(len(results), 4)
|
|
|
|
# a parallel grid run must agree with the same signal run on its own
|
|
alone = backtest.optimize(hold_lots, lots=[0.2], unused=["a"])
|
|
matching = [s for p, s in results if p == {"lots": 0.2, "unused": "a"}]
|
|
self.assertEqual(matching[0].final_cash, alone[0][1].final_cash)
|
|
|
|
def test_maximize_picks_the_named_field(self):
|
|
def hold_lots(df, lots):
|
|
return [lots] * len(df)
|
|
|
|
backtest = Backtest(rising_market(), cash=10_000.0)
|
|
results = backtest.optimize(hold_lots, maximize="max_drawdown_pct", lots=[0.1, 1.0])
|
|
|
|
self.assertEqual(
|
|
[stats.max_drawdown_pct for _, stats in results],
|
|
sorted([stats.max_drawdown_pct for _, stats in results], reverse=True),
|
|
)
|
|
|
|
def test_nan_signal_is_rejected_rather_than_silently_held(self):
|
|
def leaky_warmup(df, lots):
|
|
return [float("nan")] + [lots] * (len(df) - 1)
|
|
|
|
backtest = Backtest(rising_market(), cash=10_000.0)
|
|
with self.assertRaisesRegex(ValueError, "NaN"):
|
|
backtest.optimize(leaky_warmup, lots=[0.1])
|
|
|
|
def test_wrong_length_signal_is_rejected(self):
|
|
backtest = Backtest(rising_market(), cash=10_000.0)
|
|
with self.assertRaisesRegex(ValueError, "one per bar"):
|
|
backtest.optimize(lambda df, lots: [lots] * 3, lots=[0.1])
|
|
|
|
def test_empty_grid_is_rejected(self):
|
|
backtest = Backtest(rising_market(), cash=10_000.0)
|
|
with self.assertRaises(ValueError):
|
|
backtest.optimize(lambda df: [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|