Deepseek- 指标和自定义策略方法生成买卖点
来源: Trilium
你是一名A股量化交易专家，请协助分析，在利用指标算法（如MACD，RSI等）对自选股池进行计算时，如何根据这些指标和自定义策略方法生成买卖点提示？对于已生成买卖点的，如何进行回测分析？以分析和评价该策略的有效性？请使用akshare获取股票数据，并用ta-lib计算相关指标，用python给出相关代码，谢谢
A股量化交易策略分析与回测
作为A股量化交易专家，我将为您详细介绍如何利用技术指标生成买卖点，并进行策略回测分析。
策略构建流程
- 数据获取：使用akshare获取A股历史数据
- 指标计算：使用TA-Lib计算技术指标
- 信号生成：基于指标组合生成买卖信号
- 回测分析：评估策略表现
完整代码实现
import akshare as ak
import talib
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from tqdm import tqdm
# 设置中文显示
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False
def fetch_stock_data(stock_code, start_date, end_date):
"""
获取A股历史数据
:param stock_code: 股票代码，如 '000001'
:param start_date: 开始日期，格式 'YYYYMMDD'
:param end_date: 结束日期，格式 'YYYYMMDD'
:return: DataFrame
"""
try:
# 获取后复权数据
df = ak.stock_zh_a_hist(symbol=stock_code, period="daily",
start_date=start_date, end_date=end_date,
adjust="hfq")
df.rename(columns={
'日期': 'date',
'开盘': 'open',
'收盘': 'close',
'最高': 'high',
'最低': 'low',
'成交量': 'volume'
}, inplace=True)
df['date'] = pd.to_datetime(df['date'])
df.set_index('date', inplace=True)
df.sort_index(inplace=True)
return df
except Exception as e:
print(f"获取数据失败: {e}")
return None
def calculate_technical_indicators(df):
"""
计算技术指标
:param df: 包含OHLCV数据的DataFrame
:return: 添加了技术指标的DataFrame
"""
# 计算MACD
df['macd'], df['macd_signal'], df['macd_hist'] = talib.MACD(
df['close'], fastperiod=12, slowperiod=26, signalperiod=9)
# 计算RSI
df['rsi'] = talib.RSI(df['close'], timeperiod=14)
# 计算布林带
df['upper_band'], df['middle_band'], df['lower_band'] = talib.BBANDS(
df['close'], timeperiod=20, nbdevup=2, nbdevdn=2, matype=0)
# 计算KDJ
df['slowk'], df['slowd'] = talib.STOCH(
df['high'], df['low'], df['close'],
fastk_period=9, slowk_period=3, slowk_matype=0,
slowd_period=3, slowd_matype=0)
# 计算均线
df['ma5'] = talib.MA(df['close'], timeperiod=5)
df['ma10'] = talib.MA(df['close'], timeperiod=10)
df['ma20'] = talib.MA(df['close'], timeperiod=20)
return df
def generate_trading_signals(df):
"""
生成交易信号
:param df: 包含技术指标的DataFrame
:return: 添加了交易信号的DataFrame
"""
# 初始化信号列
df['signal'] = 0 # 0表示无信号，1表示买入，-1表示卖出
# MACD策略：金叉买入，死叉卖出
df['macd_signal'] = 0
df.loc[(df['macd'] > df['macd_signal']) & (df['macd'].shift(1) <= df['macd_signal'].shift(1)), 'macd_signal'] = 1
df.loc[(df['macd'] < df['macd_signal']) & (df['macd'].shift(1) >= df['macd_signal'].shift(1)), 'macd_signal'] = -1
# RSI策略：超卖买入，超买卖出
df['rsi_signal'] = 0
df.loc[df['rsi'] < 30, 'rsi_signal'] = 1
df.loc[df['rsi'] > 70, 'rsi_signal'] = -1
# 均线策略：5日线上穿10日线买入，下穿卖出
df['ma_signal'] = 0
df.loc[(df['ma5'] > df['ma10']) & (df['ma5'].shift(1) <= df['ma10'].shift(1)), 'ma_signal'] = 1
df.loc[(df['ma5'] < df['ma10']) & (df['ma5'].shift(1) >= df['ma10'].shift(1)), 'ma_signal'] = -1
# 综合信号生成（可根据需要调整权重）
df['signal'] = df['macd_signal'] + df['rsi_signal'] + df['ma_signal']
# 信号过滤：只有当综合信号绝对值>=2时才产生交易信号
df['final_signal'] = 0
df.loc[df['signal'] >= 2, 'final_signal'] = 1 # 买入信号
df.loc[df['signal'] <= -2, 'final_signal'] = -1 # 卖出信号
# 避免连续买入或卖出信号
position = 0 # 0表示空仓，1表示持仓
for i in range(1, len(df)):
if position == 0 and df['final_signal'].iloc[i] == 1:
df['final_signal'].iloc[i] = 1
position = 1
elif position == 1 and df['final_signal'].iloc[i] == -1:
df['final_signal'].iloc[i] = -1
position = 0
else:
df['final_signal'].iloc[i] = 0
return df
def backtest_strategy(df, initial_capital=100000, transaction_cost=0.001):
"""
回测策略
:param df: 包含交易信号的DataFrame
:param initial_capital: 初始资金
:param transaction_cost: 交易成本（手续费+滑点）
:return: 回测结果DataFrame
"""
capital = initial_capital
position = 0 # 持仓数量
entry_price = 0 # 入场价格
# 初始化回测结果列
df['position'] = 0
df['capital'] = capital
df['returns'] = 0.0
df['total'] = capital
for i in range(1, len(df)):
current_price = df['close'].iloc[i]
# 买入信号
if df['final_signal'].iloc[i] == 1 and position == 0:
position = capital * (1 - transaction_cost) / current_price
entry_price = current_price
capital = 0
# 卖出信号
elif df['final_signal'].iloc[i] == -1 and position > 0:
capital = position * current_price * (1 - transaction_cost)
position = 0
entry_price = 0
# 计算当前资产总值
current_value = capital + position * current_price
df.loc[df.index[i], 'total'] = current_value
df.loc[df.index[i], 'position'] = position
df.loc[df.index[i], 'capital'] = capital
# 计算收益率
if i > 0:
prev_value = df['total'].iloc[i-1]
df.loc[df.index[i], 'returns'] = (current_value - prev_value) / prev_value
# 计算累计收益率
df['cumulative_returns'] = (1 + df['returns']).cumprod() - 1
# 计算基准收益率（买入持有策略）
df['benchmark_returns'] = df['close'] / df['close'].iloc[0] - 1
return df
def analyze_performance(df):
"""
分析策略表现
:param df: 包含回测结果的DataFrame
:return: 绩效指标字典
"""
# 总收益率
total_return = df['total'].iloc[-1] / df['total'].iloc[0] - 1
benchmark_return = df['benchmark_returns'].iloc[-1]
# 年化收益率
days = len(df)
years = days / 252
annualized_return = (1 + total_return) ** (1/years) - 1
annualized_benchmark = (1 + benchmark_return) ** (1/years) - 1
# 最大回撤
df['peak'] = df['total'].cummax()
df['drawdown'] = (df['peak'] - df['total']) / df['peak']
max_drawdown = df['drawdown'].max()
# 胜率
trade_returns = df[df['final_signal'] != 0]['returns']
win_rate = len(trade_returns[trade_returns > 0]) / len(trade_returns) if len(trade_returns) > 0 else 0
# 平均盈亏比
winning_trades = trade_returns[trade_returns > 0]
losing_trades = trade_returns[trade_returns <= 0]
profit_loss_ratio = abs(winning_trades.mean() / losing_trades.mean()) if len(losing_trades) > 0 else np.inf
# 夏普比率（假设无风险利率为3%）
risk_free_rate = 0.03
excess_returns = df['returns'] - risk_free_rate/252
sharpe_ratio = excess_returns.mean() / excess_returns.std() * np.sqrt(252)
return {
'总收益率': total_return,
'基准收益率': benchmark_return,
'年化收益率': annualized_return,
'基准年化收益率': annualized_benchmark,
'最大回撤': max_drawdown,
'胜率': win_rate,
'盈亏比': profit_loss_ratio,
'夏普比率': sharpe_ratio
}
def plot_results(df, stock_code):
"""
绘制回测结果
:param df: 包含回测结果的DataFrame
:param stock_code: 股票代码
"""
plt.figure(figsize=(14, 10))
# 资产曲线
plt.subplot(2, 1, 1)
plt.plot(df.index, df['total'], label='策略净值', color='b')
plt.plot(df.index, df['close'] / df['close'].iloc[0] * df['total'].iloc[0],
label='基准净值', color='g', alpha=0.7)
plt.title(f'{stock_code} 策略回测结果')
plt.ylabel('净值')
plt.legend()
plt.grid(True)
# 回撤曲线
plt.subplot(2, 1, 2)
plt.plot(df.index, df['drawdown'], label='回撤', color='r')
plt.fill_between(df.index, df['drawdown'], color='r', alpha=0.3)
plt.ylabel('回撤')
plt.xlabel('日期')
plt.legend()
plt.grid(True)
plt.tight_layout()
plt.show()
def main(stock_codes, start_date, end_date):
"""
主函数
:param stock_codes: 股票代码列表
:param start_date: 开始日期
:param end_date: 结束日期
"""
for code in tqdm(stock_codes, desc='处理股票'):
# 获取数据
df = fetch_stock_data(code, start_date, end_date)
if df is None or len(df) < 100:
print(f"股票 {code} 数据不足或获取失败，跳过")
continue
# 计算技术指标
df = calculate_technical_indicators(df)
# 生成交易信号
df = generate_trading_signals(df)
# 回测策略
df = backtest_strategy(df)
# 分析表现
performance = analyze_performance(df)
# 打印结果
print(f"\n股票 {code} 策略表现:")
for k, v in performance.items():
print(f"{k}: {v:.2%}" if isinstance(v, float) else f"{k}: {v:.2f}")
# 绘制图表
plot_results(df, code)
if __name__ == "__main__":
# 示例股票池
stock_pool = ['000001', '600000', '601318']
# 设置日期范围
start_date = '20200101'
end_date = '20231231'
# 运行主程序
main(stock_pool, start_date, end_date)
关键步骤解析
1. 数据获取
使用akshare的stock_zh_a_hist函数获取A股后复权历史数据，确保价格数据考虑了分红送股的影响。
2. 指标计算
使用TA-Lib计算了以下技术指标： - MACD（移动平均收敛发散指标） - RSI（相对强弱指数） - 布林带（Bollinger Bands） - KDJ指标 - 简单移动平均线（5日、10日、20日）
3. 信号生成
实现了三种信号生成方法并进行了综合： 1. MACD策略：金叉买入，死叉卖出 2. RSI策略：RSI<30超卖买入，RSI>70超买卖出 3. 均线策略：5日线上穿10日线买入，下穿卖出
综合信号生成后，还进行了信号过滤和避免连续信号的逻辑处理。
4. 回测分析
回测系统考虑了： - 初始资金管理 - 交易成本（手续费+滑点） - 仓位管理 - 资产总值计算
5. 绩效评估
计算了以下关键绩效指标： - 总收益率和年化收益率 - 最大回撤 - 胜率和盈亏比 - 夏普比率 - 与基准（买入持有策略）的对比
策略优化建议
- 参数优化：可以使用网格搜索或优化算法寻找最佳参数组合
- 多因子组合：加入更多技术指标或基本面因子
- 风险管理：加入止损止盈逻辑
- 多时间框架：结合日线、周线等多时间框架分析
- 机器学习：使用机器学习模型优化信号生成
这个框架提供了完整的量化交易策略开发流程，您可以根据实际需求调整指标组合和信号生成逻辑。