2. MACD多周期共振信号
任务目标
本案例实现一个基于日、周、月三个周期MACD指标的多周期共振信号系统。当三个周期的MACD柱状线同时为正（金叉状态）时产生共振买入信号，同时为负（死叉状态）时产生共振卖出信号。系统使用tushare Pro获取行情数据，计算各周期MACD后对齐到日线输出信号。
环境与依赖
需要安装：tushare、pandas、numpy。使用tushare Pro接口需在官网注册获取，并在代码中替换为你的。
核心实现
import tushare as ts
import pandas as pd
import numpy as np
# 设置并初始化接口（替换为你的）
ts.set_('your_tushare_')
pro = ts.pro_api()
def get_kline(ts_code, freq, start_date, end_date):
"""获取指定周期的K线数据"""
df = pro.daily(ts_code=ts_code, start_date=start_date, end_date=end_date)
if freq == 'W':
df = pro.weekly(ts_code=ts_code, start_date=start_date, end_date=end_date)
elif freq == 'M':
df = pro.monthly(ts_code=ts_code, start_date=start_date, end_date=end_date)
df = df.sort_values('trade_date').reset_index(drop=True)
return df
def calc_macd(df, fast=12, slow=26, signal=9):
"""计算MACD指标，返回DIF、DEA、MACD柱"""
close = df['close']
ema_fast = close.ewm(span=fast, adjust=False).mean()
ema_slow = close.ewm(span=slow, adjust=False).mean()
dif = ema_fast - ema_slow
dea = dif.ewm(span=signal, adjust=False).mean()
macd = (dif - dea) * 2
return dif, dea, macd
def multi_period_macd_signal(ts_code, start_date, end_date):
"""
生成多周期MACD共振信号
返回DataFrame：trade_date, close, daily_macd, weekly_macd, monthly_macd, signal
signal: 1=共振买入, -1=共振卖出, 0=无信号
"""
# 获取三个周期的K线
daily = get_kline(ts_code, 'D', start_date, end_date)
weekly = get_kline(ts_code, 'W', start_date, end_date)
monthly = get_kline(ts_code, 'M', start_date, end_date)
# 计算各周期MACD柱
_, _, daily_macd = calc_macd(daily)
_, _, weekly_macd = calc_macd(weekly)
_, _, monthly_macd = calc_macd(monthly)
# 将周、月MACD对齐到日线（使用最近一期值填充）
daily['daily_macd'] = daily_macd
daily['trade_date'] = pd.to_datetime(daily['trade_date'])
weekly['trade_date'] = pd.to_datetime(weekly['trade_date'])
monthly['trade_date'] = pd.to_datetime(monthly['trade_date'])
# 周线对齐：将周MACD映射到该周最后一个交易日
weekly_map = weekly.set_index('trade_date')['macd']
daily['weekly_macd'] = daily['trade_date'].map(
lambda d: weekly_map[weekly_map.index <= d].iloc[-1] if len(weekly_map[weekly_map.index <= d]) > 0 else np.nan
)
# 月线对齐：将月MACD映射到该月最后一个交易日
monthly_map = monthly.set_index('trade_date')['macd']
daily['monthly_macd'] = daily['trade_date'].map(
lambda d: monthly_map[monthly_map.index <= d].iloc[-1] if len(monthly_map[monthly_map.index <= d]) > 0 else np.nan
)
# 生成共振信号
daily['signal'] = 0
daily.loc[(daily['daily_macd'] > 0) & (daily['weekly_macd'] > 0) & (daily['monthly_macd'] > 0), 'signal'] = 1
daily.loc[(daily['daily_macd'] < 0) & (daily['weekly_macd'] < 0) & (daily['monthly_macd'] < 0), 'signal'] = -1
# 去除前几个月的NaN值（因周/月MACD需要足够数据计算）
result = daily.dropna(subset=['weekly_macd', 'monthly_macd']).reset_index(drop=True)
return result[['trade_date', 'close', 'daily_macd', 'weekly_macd', 'monthly_macd', 'signal']]
# 使用示例
if __name__ == '__main__':
# 以贵州茅台为例，获取2023年数据
df = multi_period_macd_signal('600519.SH', '20230101', '20231231')
print(df.head(10))
# 统计信号数量
buy_signals = df[df['signal'] == 1]
sell_signals = df[df['signal'] == -1]
print(f"\n共振买入信号次数: {len(buy_signals)}")
print(f"共振卖出信号次数: {len(sell_signals)}")
# 显示最近一次买入信号日期
if len(buy_signals) > 0:
print(f"最近买入信号日期: {buy_signals.iloc[-1]['trade_date'].date()}")
运行与验证
运行方式：将上述代码保存为 macd_resonance.py，替换后直接运行：
python macd_resonance.py验证方法：程序会输出前10行数据，包含各周期MACD值和信号列。手动检查：
- 当
signal=1时，确认该行三个MACD值均大于0； - 当
signal=-1时，确认该行三个MACD值均小于0； - 信号数量统计合理（通常一年内共振信号出现次数有限，约几次到十几次）。
预期输出示例（前几行，具体数值取决于行情）：
trade_date close daily_macd weekly_macd monthly_macd signal
0 2023-01-03 1755.0000 0.123456 0.234567 0.345678 1
1 2023-01-04 1760.0000 0.234567 0.234567 0.345678 1
...关键说明
- 周、月MACD对齐采用“最近一期已完成周期”的映射方式，即日线数据使用截至当日最近的周/月MACD值，避免未来函数。
- MACD参数使用默认值（12, 26, 9），可根据需要调整。
- 由于周/月MACD需要至少26周/月数据才能稳定，建议回测起始日期至少提前半年以上，否则前期信号可能不准确。
- 本实现仅输出信号，不包含交易策略或回测逻辑，如需扩展可在信号基础上自行添加持仓判断。