量化策略代码信号生成异常：条件失效及无止损止盈信号排查

问题概述 #

1. 移动平均线策略中，移除买入信号的(self.df['Accu_Position'] == 0)持仓过滤条件后，回测结果完全一致，说明该条件未实际生效。
2. 策略全程未生成任何止损（StopLoss）、止盈（TakeProfit）信号，触发逻辑失效。

完整策略代码：

from matplotlib import pylab
import numpy as np
import pandas as pd
from backtesting import Backtest , Strategy
import matplotlib.pyplot as plt

df = pd.read_excel("E:/usdcnh.xls")
print(df.columns)

class MovingAverageStrategy:
    def __init__(self, df, window_size, buy_threshold, sell_threshold, stop_loss, take_profit):
        self.df = df
        self.window_size = window_size
        self.buy_threshold = buy_threshold
        self.sell_threshold = sell_threshold

        self.stop_loss = stop_loss
        self.take_profit = take_profit

    def calculate_moving_average(self):
        self.df['MovingAverage'] = self.df['Close'].rolling(window=self.window_size).mean()

    def calculate_deviation(self):
        self.df['Deviation'] = self.df['Close'] - self.df['MovingAverage']

    def generate_signals(self):
        self.calculate_moving_average()
        self.calculate_deviation()

        self.df['Position'] = 0 
        self.df['Accu_Position'] = self.df['Position'] + self.df['Position'].shift(-1)
        self.df['Accu_Position'].iloc[-1] = self.df['Position'].iloc[-1]

        self.df['Buy_Condition'] = (self.df['Deviation'] <= self.buy_threshold) & (self.df['Accu_Position'] == 0)
        self.df['Buy_Signal'] = np.where(self.df['Buy_Condition'], 1, 0)
        self.df['Sell_Condition'] = (self.df['Deviation'] >= self.sell_threshold) & (self.df['Accu_Position'] == 0)
        self.df['Sell_Signal'] = np.where(self.df['Sell_Condition'] , -1, 0)

        self.df['EntryPrice'] = np.where((self.df['Buy_Signal']!=0) | (self.df['Sell_Signal']!=0), self.df['Close'], 0)
        self.df['EntryPriceRecord'] = np.where(self.df['EntryPrice']!=0, self.df['EntryPrice'], self.df['EntryPrice'].shift(-1)) 
        self.df['PriceChange'] = self.df['Close'] - self.df['EntryPriceRecord']

        self.df['StopLoss_Signal'] = 0  
        stop_loss_condition = ((self.df['Accu_Position'] == 1) & (self.df['PriceChange'] <= -self.stop_loss)) | ((self.df['Accu_Position'] == -1) & (self.df['PriceChange'] >= self.stop_loss))
        self.df.loc[stop_loss_condition, 'StopLoss_Signal'] = -self.df['Accu_Position']

        self.df['TakeProfit_Signal'] = 0
        take_profit_condition = ((self.df['Accu_Position'] == 1) & (self.df['PriceChange'] >= self.take_profit)) | ((self.df['Accu_Position'] == -1) & (self.df['PriceChange'] <= -self.take_profit))
        self.df.loc[take_profit_condition, 'TakeProfit_Signal'] = -self.df['Accu_Position']

        self.df['Signal'] = self.df[['Buy_Signal', 'Sell_Signal', 'StopLoss_Signal', 'TakeProfit_Signal']].sum(axis=1)
        self.df['Position'] += self.df['Signal']

        num_buy_signals = self.df['Buy_Signal'].sum()
        num_sell_signals = self.df['Sell_Signal'].sum()
        num_take_profit_signals = self.df['TakeProfit_Signal'].sum()
        num_stop_loss_signals = self.df['StopLoss_Signal'].sum()

        print('Number of Buy Signals:', num_buy_signals)
        print('Number of Sell Signals:', num_sell_signals)
        print('Number of Take Profit Signals:', num_take_profit_signals)
        print('Number of Stop Loss Signals:', num_stop_loss_signals)
        print('Number of Stop Loss Signals:', num_stop_loss_signals)
        print(self.df['Accu_Position'])

    def calculate_pnl(self):
        self.df['PnL'] = self.df['Close'] * self.df['Position']
        self.df['AccuPnl'] = self.df['PnL'] + self.df['PnL'].shift()
        print('Total PnL:', self.df['PnL'].sum())


    def apply_strategy(self):
        self.calculate_moving_average()
        self.calculate_deviation()
        self.generate_signals()
        self.calculate_pnl()
        return self.df

strategy = MovingAverageStrategy(df, window_size=200, buy_threshold=-0.0350, sell_threshold=0.0350, stop_loss=0.0020, take_profit=0.0050)
df = strategy.apply_strategy()

# # Plotting the equity
plt.figure(figsize=(10, 6))
plt.plot(df['Close'], label='Equity')
plt.plot(df['MovingAverage'], label='MovingAverage')

# # Plotting the buy and sell points
buy_points = df[df['Buy_Signal'] != 0]
sell_points = df[df['Sell_Signal'] != 0]
StopLoss_points = df[df['StopLoss_Signal'] != 0]
TakeProfit_points = df[df['TakeProfit_Signal'] != 0]

plt.scatter(buy_points.index, buy_points['Close'], color='green', label='Buy')
plt.scatter(sell_points.index, sell_points['Close'], color='red', label='Sell')
plt.scatter(StopLoss_points.index, StopLoss_points['Close'], color='blue', label='StopLoss')
plt.scatter(TakeProfit_points.index, TakeProfit_points['Close'], color='orange', label='TakeProfit')

plt.xlabel('Date')
plt.ylabel('Equity')
plt.title('Equity and Buy/Sell Points')
plt.legend()
plt.show()

问题根源及修正方案 #

1. 持仓过滤条件无效的原因 #

Accu_Position的初始化逻辑完全错误：

self.df['Position'] = 0 
self.df['Accu_Position'] = self.df['Position'] + self.df['Position'].shift(-1)

初始Position全为0，导致Accu_Position也全为0，(self.df['Accu_Position'] == 0)条件永远为真，移除与否自然不影响结果。

修正方式：
Accu_Position是持仓累计值，需基于生成的Signal逐步计算，放在信号生成之后：

# 先计算所有信号总和
self.df['Signal'] = self.df[['Buy_Signal', 'Sell_Signal', 'StopLoss_Signal', 'TakeProfit_Signal']].sum(axis=1)
# 累计得到持仓，限制单仓范围为-1/0/1
self.df['Position'] = self.df['Signal'].cumsum().clip(-1, 1)
# Accu_Position直接复用正确的持仓值
self.df['Accu_Position'] = self.df['Position']

2. 止损止盈无信号的双重原因 #

• Accu_Position全为0，导致止损止盈的持仓判断条件永远不成立；
• EntryPriceRecord用shift(-1)取后一行值，无法持续填充持仓期间的入场价，导致PriceChange计算错误。

修正EntryPriceRecord逻辑：

self.df['EntryPrice'] = np.where((self.df['Buy_Signal']!=0) | (self.df['Sell_Signal']!=0), self.df['Close'], np.nan)
# 用前向填充保留持仓期间的入场价
self.df['EntryPriceRecord'] = self.df['EntryPrice'].ffill().fillna(0)

修正后的完整generate_signals方法 #

def generate_signals(self):
    self.calculate_moving_average()
    self.calculate_deviation()

    # 初始化所有信号列
    self.df['Position'] = 0 
    self.df['Buy_Signal'] = 0
    self.df['Sell_Signal'] = 0
    self.df['StopLoss_Signal'] = 0
    self.df['TakeProfit_Signal'] = 0
    self.df['Accu_Position'] = 0

    # 生成买卖信号（基于初始空仓条件）
    buy_condition = (self.df['Deviation'] <= self.buy_threshold) & (self.df['Accu_Position'] == 0)
    self.df.loc[buy_condition, 'Buy_Signal'] = 1

    sell_condition = (self.df['Deviation'] >= self.sell_threshold) & (self.df['Accu_Position'] == 0)
    self.df.loc[sell_condition, 'Sell_Signal'] = -1

    # 计算入场价并填充持仓期间的值
    self.df['EntryPrice'] = np.where((self.df['Buy_Signal']!=0) | (self.df['Sell_Signal']!=0), self.df['Close'], np.nan)
    self.df['EntryPriceRecord'] = self.df['EntryPrice'].ffill().fillna(0)
    self.df['PriceChange'] = self.df['Close'] - self.df['EntryPriceRecord']

    # 计算初始持仓
    self.df['Signal'] = self.df[['Buy_Signal', 'Sell_Signal', 'StopLoss_Signal', 'TakeProfit_Signal']].sum(axis=1)
    self.df['Position'] = self.df['Signal'].cumsum().clip(-1, 1)
    self.df['Accu_Position'] = self.df['Position']

    # 基于正确持仓生成止损止盈信号
    # 多头止损
    long_stop = (self.df['Accu_Position'] == 1) & (self.df['PriceChange'] <= -self.stop_loss)
    self.df.loc[long_stop, 'StopLoss_Signal'] = -1
    # 空头止损
    short_stop = (self.df['Accu_Position'] == -1) & (self.df['PriceChange'] >= self.stop_loss)
    self.df.loc[short_stop, 'StopLoss_Signal'] = 1

    # 多头止盈
    long_tp = (self.df['Accu_Position'] == 1) & (self.df['PriceChange'] >= self.take_profit)
    self.df.loc[long_tp, 'TakeProfit_Signal'] = -1
    # 空头止盈
    short_tp = (self.df['Accu_Position'] == -1) & (self.df['PriceChange'] <= -self.take_profit)
    self.df.loc[short_tp, 'TakeProfit_Signal'] = 1

    # 重新计算总信号和最终持仓
    self.df['Signal'] = self.df[['Buy_Signal', 'Sell_Signal', 'StopLoss_Signal', 'TakeProfit_Signal']].sum(axis=1)
    self.df['Position'] = self.df['Signal'].cumsum().clip(-1, 1)
    self.df['Accu_Position'] = self.df['Position']

    # 统计信号数量
    num_buy_signals = self.df['Buy_Signal'].sum()
    num_sell_signals = self.df['Sell_Signal'].sum()
    num_take_profit_signals = self.df['TakeProfit_Signal'].sum()
    num_stop_loss_signals = self.df['StopLoss_Signal'].sum()

    print('Number of Buy Signals:', num_buy_signals)
    print('Number of Sell Signals:', num_sell_signals)
    print('Number of Take Profit Signals:', num_take_profit_signals)
    print('Number of Stop Loss Signals:', num_stop_loss_signals)
    print(self.df['Accu_Position'])

额外优化：盈亏计算修正 #

原calculate_pnl逻辑错误，正确的每日盈亏应基于持仓和价格变动：

def calculate_pnl(self):
    # 每日盈亏 = (当日收盘价 - 前日收盘价) * 当日持仓
    self.df['PnL'] = (self.df['Close'] - self.df['Close'].shift(1)) * self.df['Position']
    # 累计盈亏
    self.df['AccuPnl'] = self.df['PnL'].cumsum()
    print('Total PnL:', self.df['AccuPnl'].iloc[-1])

内容的提问来源于stack exchange，提问作者Postman1999