""" 信号生成模块 """ import numpy as np import pandas as pd from typing import Optional, TYPE_CHECKING if TYPE_CHECKING: from pandas import Series def generate_signals( score: 'pd.Series', buy_threshold: float = 0.8, sell_threshold: float = -0.8, window: int = 30, use_rolling_std: bool = True ) -> 'pd.Series': """ 基于因子得分生成买卖信号 Parameters: ----------- score : Series 因子综合得分 buy_threshold : float 买入阈值(标准差倍数) sell_threshold : float 卖出阈值(标准差倍数) window : int 滚动窗口(用于计算标准差) use_rolling_std : bool 是否使用滚动标准差 Returns: -------- Series: 交易信号(1=买入,-1=卖出,0=持有) """ signals = pd.Series(0, index=score.index) if use_rolling_std: # 使用滚动标准差 rolling_std = score.rolling(window).std() buy_line = buy_threshold * rolling_std sell_line = sell_threshold * rolling_std else: # 使用固定阈值 std = score.std() buy_line = buy_threshold * std sell_line = sell_threshold * std # 生成原始信号 raw_signals = pd.Series(0, index=score.index) raw_signals[score > buy_line] = 1 # 买入信号 raw_signals[score < sell_line] = -1 # 卖出信号 # 只在信号变化时产生交易信号,其他时候保持持仓状态 signals = pd.Series(0, index=score.index) position = 0 # 当前持仓状态:0=空仓,1=满仓 for i in range(len(raw_signals)): current_signal = raw_signals.iloc[i] # 只在信号变化时产生交易 if current_signal == 1 and position == 0: signals.iloc[i] = 1 # 买入 position = 1 elif current_signal == -1 and position == 1: signals.iloc[i] = -1 # 卖出 position = 0 # 其他情况保持当前持仓状态,不产生交易信号 return signals.astype(int) def generate_signals_with_position( score: 'pd.Series', buy_threshold: float = 0.8, sell_threshold: float = -0.8, window: int = 30, current_position: int = 0 ) -> 'pd.Series': """ 生成信号(考虑当前持仓状态) Parameters: ----------- current_position : int 当前持仓:0=空仓,1=满仓 """ raw_signals = generate_signals(score, buy_threshold, sell_threshold, window) signals = pd.Series(0, index=score.index) position = current_position for i in range(len(raw_signals)): signal = raw_signals.iloc[i] if signal == 1 and position == 0: signals.iloc[i] = 1 # 买入 position = 1 elif signal == -1 and position == 1: signals.iloc[i] = -1 # 卖出 position = 0 else: signals.iloc[i] = 0 # 持有 return signals