fix(backtest): strategy position not tracked during signal generation

Root cause: _generate_signals() iterated all bars calling strategy.next()
but never updated _position_size or _cash on the strategy. Strategies
that check self.position['size'] before buy/sell always saw 0, producing
only BUY signals with no SELL — exhausting cash and producing drawdowns
exceeding 100%.

Fix: add _update_strategy_position() that estimates position changes
after each bar's signals using close price. This gives the strategy an
accurate view of its holdings so it can correctly alternate buy/sell.

Regression tests added:
- test_position_aware_buy_sell_alternation: verifies BUY/SELL alternation
- test_position_aware_no_duplicate_buys: no suspicious tiny duplicate buys

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
GitHub
2026-06-09 18:50:13 +08:00
co-authored by Claude Opus 4.8
parent f7e1abd873
commit 6a6d75f5d5
2 changed files with 130 additions and 4 deletions
+57 -4
View File
@@ -137,6 +137,11 @@ class BacktestEngine:
def _generate_signals(self, df: pd.DataFrame, chanlun_result: Any | None) -> list[Signal]:
"""Generate signals from strategy.
After each bar's signals, update strategy's internal position state so
subsequent bars can make informed decisions (e.g., "don't buy if already
holding"). Uses close price as estimate; actual execution price is
determined later by OrderSimulator.
Args:
df: Price data
chanlun_result: Optional chanlun analysis result
@@ -156,6 +161,11 @@ class BacktestEngine:
if chanlun_result is not None:
strat._chanlun_result = chanlun_result
# Initialize position tracking
strat._cash = self._cash
strat._position_size = 0.0
close_arr = df["close"].to_numpy()
# Call init
strat._call_init()
@@ -165,10 +175,48 @@ class BacktestEngine:
strat._set_bar_index(i)
strat._call_next()
bar_signals = strat._clear_signals()
# Update strategy position state so next bar sees current holdings
self._update_strategy_position(strat, bar_signals, close_arr[i])
all_signals.extend(bar_signals)
return all_signals
def _update_strategy_position(
self, strat: Strategy, signals: list[Signal], est_price: float
) -> None:
"""Update strategy's internal position estimate after each bar.
Uses close price as estimate for full-position calculations.
The actual execution price is determined by OrderSimulator later.
Args:
strat: Strategy instance
signals: Signals generated on this bar
est_price: Estimated price (close of current bar)
"""
for sig in signals:
price = sig.price or est_price
if sig.direction == "BUY":
if sig.size == 0:
# Full position: estimate shares (100-lot rounding)
shares = int(strat._cash / (price * (1 + self._commission)) / 100) * 100
if shares > 0:
strat._position_size += shares
strat._cash -= shares * price
else:
strat._position_size += sig.size
strat._cash -= sig.size * price
elif sig.direction == "SELL":
if sig.size == 0:
# Full sell
strat._cash += strat._position_size * price
strat._position_size = 0.0
else:
strat._cash += sig.size * price
strat._position_size = max(0.0, strat._position_size - sig.size)
def _compute_pnls(self, trades: list[Trade]) -> list[Trade]:
"""Compute realized PnL for sell trades.
@@ -191,10 +239,10 @@ class BacktestEngine:
if position_size > 0:
avg_cost = position_cost / position_size
trade.pnl = (trade.price - avg_cost) * trade.size - trade.commission
position_cost -= avg_cost * trade.size
position_size -= trade.size
else:
trade.pnl = 0.0
position_cost -= avg_cost * trade.size
position_size -= trade.size
return trades
@@ -240,10 +288,15 @@ class BacktestEngine:
Returns:
BacktestResult with empty DataFrames
"""
perf = PerformanceAnalyzer(
pd.DataFrame(columns=["total", "drawdown"]),
pd.DataFrame(columns=["direction", "pnl", "rejected"]),
).compute()
return BacktestResult(
performance={},
performance=perf,
equity_curve=pd.DataFrame(
columns=["datetime", "cash", "position_value", "total", "drawdown"]
columns=["datetime", "cash", "position_value", "total", "drawdown", "drawdown_pct"]
),
trades=pd.DataFrame(
columns=[
+73
View File
@@ -286,3 +286,76 @@ def test_pnl_calculation():
# For buy trades, PnL should be 0
assert (buy_trades["pnl"] == 0).all()
class PositionAwareStrategy(Strategy):
"""Strategy that checks position before trading (the common pattern)."""
def init(self):
self.ma5 = self.I(MyTT.MA, self.data.close, 5)
self.ma20 = self.I(MyTT.MA, self.data.close, 20)
self.cross_up = False
self.cross_down = False
def next(self):
if self._bar_index > 0:
prev_ma5 = self.ma5[self._bar_index - 1]
prev_ma20 = self.ma20[self._bar_index - 1]
curr_ma5 = self.ma5[self._bar_index]
curr_ma20 = self.ma20[self._bar_index]
if prev_ma5 <= prev_ma20 and curr_ma5 > curr_ma20:
if self.position["size"] == 0:
self.buy(size=0)
elif prev_ma5 >= prev_ma20 and curr_ma5 < curr_ma20:
if self.position["size"] > 0:
self.sell(size=0)
def test_position_aware_buy_sell_alternation():
"""Regression: strategy that checks position must produce alternating BUY/SELL."""
df = _make_df(n=300, seed=42)
engine = BacktestEngine(PositionAwareStrategy, cash=100000)
result = engine.run(df)
trades = result.trades[result.trades["rejected"] == False]
directions = trades["direction"].tolist()
# Must have both BUYs and SELLs
assert "BUY" in directions, "No BUY trades generated"
assert "SELL" in directions, "No SELL trades generated — position feedback broken"
# Trades must alternate: no two consecutive BUYs or SELLs
for i in range(1, len(directions)):
assert directions[i] != directions[i - 1], (
f"Consecutive same-direction trades at index {i}: "
f"{directions[i - 1]} -> {directions[i]}"
)
def test_position_aware_no_duplicate_buys():
"""After a BUY, position['size'] > 0 so strategy should not buy again."""
df = _make_df(n=300, seed=42)
engine = BacktestEngine(PositionAwareStrategy, cash=100000)
result = engine.run(df)
buy_trades = result.trades[
(result.trades["direction"] == "BUY") & (result.trades["rejected"] == False)
]
# Each BUY's size should be reasonable (not tiny leftover from exhausted cash)
if len(buy_trades) > 1:
# No consecutive buys where the second is tiny (cash leftover artifact)
sizes = buy_trades["size"].tolist()
for i in range(1, len(sizes)):
# Second buy in a pair should not be tiny compared to first
# (would indicate position wasn't tracked between bars)
if i >= 1:
prev_size = sizes[i - 1]
cur_size = sizes[i]
# Allow some variance but not orders-of-magnitude difference
if prev_size > 0:
assert cur_size > prev_size * 0.1, (
f"Suspicious tiny buy {cur_size} after {prev_size}"
f"position feedback may be broken"
)