快速傅里叶变换 (FFT) 分析股价波动 Python 示例核心思路股价原始序列是时域信号通过 FFT 转换到频域分离长期趋势、中期波动、短期噪声再逆变换还原平滑后的股价实现降噪、周期提取。下面示例包含数据模拟 / 真实股价获取、FFT 频域分析、滤波降噪、绘图可视化。下面给出滚动窗口 FFT 周期分析 自动生成买卖信号完整版直接用 A 股真实行情可直接跑。特点滑动窗口滚动傅里叶变换捕捉实时变化的周期提取主周期计算相位给出买入/卖出信号可视化股价、平滑线、买卖点、周期强度1. 先安装依赖pipinstallnumpy pandas matplotlib tushare2. 完整可运行代码importnumpyasnpimportpandasaspdimportmatplotlib.pyplotaspltimporttushareasts# ---------------------- 配置Tushare ----------------------ts.set_token(你的Tushare Token)prots.pro_api()# ---------------------- 参数设置 ----------------------STOCK_CODE300502.SZ# 股票代码START_DATE20240101END_DATE20260531WINDOW120# 滚动窗口大小(交易日)KEEP_NUM20# FFT保留低频分量TRADE_SIGNAL_THRESHOLD0.2# 相位阈值# ---------------------- 下载数据 ----------------------dfpro.daily(ts_codeSTOCK_CODE,start_dateSTART_DATE,end_dateEND_DATE)dfdf.sort_values(trade_date).reset_index(dropTrue)df[trade_date]pd.to_datetime(df[trade_date])pricedf[close].values datesdf[trade_date].values nlen(price)print(df.head())# 用于保存结果smooth_listnp.full(n,np.nan)signal_buynp.full(n,np.nan)signal_sellnp.full(n,np.nan)period_listnp.full(n,np.nan)# ---------------------- 滚动窗口FFT ----------------------foriinrange(WINDOW,n):win_dataprice[i-WINDOW:i]fft_valsnp.fft.fft(win_data)fft_freqnp.fft.fftfreq(WINDOW,d1)ampnp.abs(fft_vals)# 滤波fft_filteredfft_vals.copy()fft_filtered[KEEP_NUM:WINDOW-KEEP_NUM]0smoothnp.fft.ifft(fft_filtered).real[-1]smooth_list[i]smooth# 找主周期去掉0频amp[0]0main_idxnp.argmax(amp)main_freqfft_freq[main_idx]ifmain_freq0:main_period1/main_freq period_list[i]main_periodelse:main_periodNone# 相位判断买卖正弦波 0附近买入π附近卖出phasenp.angle(fft_vals[main_idx])ifmain_periodandabs(phase)TRADE_SIGNAL_THRESHOLD:signal_buy[i]price[i]elifmain_periodandabs(abs(phase)-np.pi)TRADE_SIGNAL_THRESHOLD:signal_sell[i]price[i]# ---------------------- 绘图 ----------------------plt.rcParams[font.sans-serif][SimHei]plt.rcParams[axes.unicode_minus]Falsefig,axesplt.subplots(3,1,figsize(16,12))# 图1股价FFT平滑买卖点axes[0].plot(dates,price,label收盘价,alpha0.7)axes[0].plot(dates,smooth_list,label滚动FFT平滑线,cred,lw2)axes[0].scatter(dates,signal_buy,marker^,cgreen,s80,label买入)axes[0].scatter(dates,signal_sell,markerv,cred,s80,label卖出)axes[0].set_title(f{STOCK_CODE}滚动FFT周期交易信号)axes[0].legend()axes[0].grid(True)# 图2实时主周期变化axes[1].plot(dates,period_list,cblue)axes[1].set_title(滚动窗口识别的主波动周期(交易日))axes[1].set_ylabel(周期(天))axes[1].grid(True)# 图3FFT振幅示例最后一个窗口win_dataprice[-WINDOW:]ampnp.abs(np.fft.fft(win_data))freqnp.fft.fftfreq(WINDOW,1)maskfreq0axes[2].plot(freq[mask],amp[mask])axes[2].set_title(最新窗口频域振幅谱)axes[2].set_xlabel(频率(1/天))axes[2].grid(True)plt.tight_layout()plt.show()# ---------------------- 输出统计 ----------------------valid_periodsperiod_list[~np.isnan(period_list)]print(*50)print(f股票{STOCK_CODE}分析结果)print(f平均主波动周期{np.mean(valid_periods):.1f}天)print(f最大周期{np.max(valid_periods):.1f}天)print(f最小周期{np.min(valid_periods):.1f}天)print(*50)3. 关键参数解释直接调参即可实战WINDOW120滚动窗口越大看大周期越小抓短线周期KEEP_NUM20FFT保留低频分量越大波动越多越小越平滑TRADE_SIGNAL_THRESHOLD0.2相位阈值越小买卖信号越严格4. 实战用法替换你的Tushare Token更换STOCK_CODE即可分析任意A股输出实时平滑股价自动绿三角买入、红倒三角卖出实时变化的主周期比如20/30/60天下面直接给出滚动FFT周期策略 完整回测完整版包含买卖信号、持仓、净值曲线、胜率、盈亏比、最大回撤、年化收益一键出结果。安装依赖pipinstallnumpy pandas matplotlib tushare完整可运行代码FFT周期策略 回测importnumpyasnpimportpandasaspdimportmatplotlib.pyplotaspltimporttushareasts# ---------------------- 配置 ----------------------ts.set_token(你的Tushare Token)prots.pro_api()# 策略参数STOCK_CODE300308.SZ# 可替换START_DATE20240101END_DATE20260531WINDOW120# FFT滚动窗口KEEP_NUM20# 保留低频分量PHASE_THRESH0.2# 相位阈值越小信号越少越稳INIT_CAPITAL100000# 初始资金FEE_RATE0.0003# 交易手续费滑点# ---------------------- 下载数据 ----------------------dfpro.daily(ts_codeSTOCK_CODE,start_dateSTART_DATE,end_dateEND_DATE)dfdf.sort_values(trade_date).reset_index(dropTrue)df[trade_date]pd.to_datetime(df[trade_date])pricedf[close].values datesdf[trade_date].values nlen(price)print(df.tail())# 信号存储smooth_listnp.full(n,np.nan)buy_signalnp.full(n,False)sell_signalnp.full(n,False)period_listnp.full(n,np.nan)# ---------------------- 滚动FFT生成买卖信号 ----------------------foriinrange(WINDOW,n):win_dataprice[i-WINDOW:i]fft_valsnp.fft.fft(win_data)fft_freqnp.fft.fftfreq(WINDOW,d1)ampnp.abs(fft_vals)amp[0]0# 去除直流趋势# 滤波平滑fft_filteredfft_vals.copy()fft_filtered[KEEP_NUM:WINDOW-KEEP_NUM]0smoothnp.fft.ifft(fft_filtered).real[-1]smooth_list[i]smooth# 主周期main_idxnp.argmax(amp)main_freqfft_freq[main_idx]ifmain_freq0:period_list[i]1/main_freqelse:continue# 相位判断接近0买入接近π卖出phasenp.angle(fft_vals[main_idx])ifabs(phase)PHASE_THRESH:buy_signal[i]Trueelifabs(abs(phase)-np.pi)PHASE_THRESH:sell_signal[i]True# ---------------------- 回测计算 ----------------------cashINIT_CAPITAL shares0net_value[]position[]trade_records[]foriinrange(n):pprice[i]# 买入ifbuy_signal[i]andshares0:sharescash/p*(1-FEE_RATE)cash0trade_records.append({date:dates[i],type:buy,price:p})# 卖出elifsell_signal[i]andshares0:cashshares*p*(1-FEE_RATE)shares0trade_records.append({date:dates[i],type:sell,price:p})nvcashshares*p net_value.append(nv)position.append(1ifshares0else0)net_valuenp.array(net_value)positionnp.array(position)# 回测指标final_capnet_value[-1]total_return(final_cap/INIT_CAPITAL)-1days(dates[-1]-dates[0]).astype(int)annual_return(1total_return)**(365/days)-1# 最大回撤peaknp.maximum.accumulate(net_value)drawdown(net_value-peak)/peak max_dddrawdown.min()# 胜率tradespd.DataFrame(trade_records)win_count0profit_sum0loss_sum0foriinrange(0,len(trades)-1,2):ifi1len(trades):breakbuy_ptrades.loc[i,price]sell_ptrades.loc[i1,price]ret(sell_p/buy_p)-1ifret0:win_count1profit_sumretelse:loss_sumret total_tradeswin_count(len(trades)//2-win_count)win_ratewin_count/total_tradesiftotal_trades0else0profit_loss_ratioabs(profit_sum/loss_sum)ifloss_sum!0elsenp.inf# ---------------------- 输出回测结果 ----------------------print(*60)print(f【FFT周期策略回测结果】{STOCK_CODE})print(f初始资金:{INIT_CAPITAL:.0f})print(f期末资金:{final_cap:.2f})print(f总收益率:{total_return:.2%})print(f年化收益率:{annual_return:.2%})print(f最大回撤:{max_dd:.2%})print(f交易次数:{total_trades})print(f胜率:{win_rate:.2%})print(f盈亏比:{profit_loss_ratio:.2f})print(*60)# ---------------------- 绘图 ----------------------plt.rcParams[font.sans-serif][SimHei]plt.rcParams[axes.unicode_minus]Falsefig,axesplt.subplots(3,1,figsize(16,12))# 图1股价平滑线买卖点axes[0].plot(dates,price,label收盘价,alpha0.7)axes[0].plot(dates,smooth_list,labelFFT平滑线,cred,lw2)axes[0].scatter(dates[buy_signal],price[buy_signal],marker^,cg,s80,label买入)axes[0].scatter(dates[sell_signal],price[sell_signal],markerv,cr,s80,label卖出)axes[0].set_title(f{STOCK_CODE}FFT周期交易信号)axes[0].legend()axes[0].grid(True)# 图2净值曲线axes[1].plot(dates,net_value,cpurple,label策略净值)axes[1].axhline(yINIT_CAPITAL,cgray,ls--,label初始资金)axes[1].set_title(策略净值曲线)axes[1].legend()axes[1].grid(True)# 图3主周期变化axes[2].plot(dates,period_list,cblue)axes[2].set_title(滚动窗口识别的主波动周期(交易日))axes[2].set_ylabel(周期(天))axes[2].grid(True)plt.tight_layout()plt.show()输出指标说明总收益率 / 年化收益率策略整体收益最大回撤账户最大亏损幅度越小越稳胜率盈利交易占比盈亏比平均盈利 / 平均亏损1.5 算优秀调参实战建议直接改这4个WINDOW120窗口越大抓大周期窗口60做短线KEEP_NUM20越大保留更多波动信号变多越小越平滑PHASE_THRESH0.2阈值越小买卖点越少过滤假信号STOCK_CODE换股票测试
