学习目标- [ ] Quant-for-Beginners · 中文零基础量化金融17天速成---课程连接涉嫌广告审核不通过一. 量化入门- 什么是量化金融 √- 你的第一个量化实验- 移动平均线策略- 策略回测二. 风险与组合- 理解波动率- 夏普比率与 Beta- 最大回撤与仓位管理- 多标的组合与相关性---学习产出problematic过去可能需要对代码基础和金融知识有一定的要求而在ai大幅发展的今天这个gap已经大幅度缩小至少在入门阶段就是如此所以这门课应运而生通常来说量化分为下面几个步骤1. 获取数据2. 分析数据3. 设计策略4. 回测策略5. 模拟交易而上面这些步骤都是可以借助ai完成 利用vibe coding生成获取数据和模拟交易的代码利用大模型分析数据和设计策略量化不是预测而是分析是先 计划你的交易 然后再 交易你的计划有的股票热门连续跌了一周有的人就会想 现在已经是谷底是我抄底的好时机 也有的股票连续涨了5天 很多人会因此追涨进去。而量化金融做的是对过去的总结而不对未来的预测。 是让你每一步的决策都有数据理论的支撑而不是一拍脑袋就决定。可视化讲解需要一定的python基础可利用ai辅助理解利用python随即生成一段股票走势图# 导入库 import numpy as np # 数值计算数组、随机数、统计 import pandas as pd # 表格数据处理像 Excel import matplotlib.pyplot as plt # 绘图库折线图、柱状图等 plt.rcParams[font.sans-serif] [SimHei] # 图表中文 plt.rcParams[axes.unicode_minus] False # 坐标轴负号正常显示 np.random.seed(42) # 固定随机数结果可复现 days 120 # 一共模拟 120 个交易日 daily_returns np.random.normal(loc0.0008, scale0.02, sizedays) # 每天随机收益率 price 100 * np.cumprod(1 daily_returns) # 从100元出发连乘成价格序列 df pd.DataFrame({ # 构建表格 收盘价: price, # 原始模拟收盘价 5日均线: pd.Series(price).rolling(5).mean(), # 最近5天均价 20日均线: pd.Series(price).rolling(20).mean() # 最近20天均价 }) # 执行本行代码 plt.figure(figsize(12, 5)) # 创建画布宽12高5英寸 plt.plot(df[收盘价], label收盘价, alpha0.6, linewidth1) # 画折线图 plt.plot(df[5日均线], label5日均线 (MA5), linewidth1.5) # 画折线图 plt.plot(df[20日均线], label20日均线 (MA20), linewidth1.5) # 画折线图 plt.fill_between(range(days), df[5日均线], df[20日均线], alpha0.1, colororange) # 两线之间浅色填充 plt.title(Python 模拟股价走势 移动平均线, fontsize14) # 设置图标题 plt.xlabel(交易日) # 横轴说明 plt.ylabel(价格) # 设置纵轴标签 plt.legend() # 图例 plt.grid(True, alpha0.3) # 浅色网格方便读数 plt.tight_layout() # 自动调整子图间距避免标签被裁切 plt.show() # 在 Notebook 里显示图片 print(f起始价格: ¥{price[0]:.2f}) # 第一天价格 print(f最终价格: ¥{price[-1]:.2f}) # 最后一天价格 print(f区间收益率: {(price[-1]/price[0] - 1)*100:.2f}%) # 整段涨跌幅 print(f最大回撤价格: ¥{price.min():.2f}) # 期间最低价 print(f最高价格: ¥{price.max():.2f}) # 期间最高价布朗运动与随机游走直观感受一下20支股票随机变化曲线这里以8%的增长作为前提这里的中位数是108.14可以看到假如你买了一支股票之后什么也不再继续操作一年后有50%概率是处于中位数之上的也就是年化8%。 问题是现实中股票市场不是8%的增长率同时你也很难找到合适的标的物。利用波动率做风险管理# 实验一布朗运动 / 随机游走 import numpy as np # 数值计算数组、随机数、统计 import matplotlib.pyplot as plt # 绘图库折线图、柱状图等 plt.rcParams[font.sans-serif] [SimHei] # 设置中文字体 plt.rcParams[axes.unicode_minus] False # 坐标轴负号正常显示 np.random.seed(2026) # 固定随机种子 n_stocks 50 # 模拟 50 只「虚拟股票」 n_days 250 # 每只走 250 个交易日约一年 start_price 100 # 起始价都是 100 元 daily_volatility 0.02 # 日波动强度标准差约2% fig, axes plt.subplots(1, 3, figsize(18, 5)) # 1行3列三个子图 # --- 左图画出 50 条价格路径 --- all_paths [] # 用来存每只股票的路径 for _ in range(n_stocks): # 循环 50 次每只股一条路径 daily_returns np.random.normal(0, daily_volatility, n_days) # 每天随机收益 price_path start_price * np.cumprod(1 daily_returns) # 价格连乘 all_paths.append(price_path) # 存入列表 axes[0].plot(price_path, alpha0.3, linewidth0.8) # 画在左图半透明 axes[0].axhline(ystart_price, colorred, linestyle--, alpha0.5, label起始价 100元) # 参考线 axes[0].set_title(50只虚拟股票的随机游走路径, fontsize13) # 设置上图标题 axes[0].set_xlabel(交易日) # 执行本行代码 axes[0].set_ylabel(价格 (元)) # 设置上图纵轴 axes[0].legend() # 显示上图图例 axes[0].grid(True, alpha0.3) # 上图显示网格 # --- 中图一年后的终点价格分布直方图--- final_prices [path[-1] for path in all_paths] # 取每条路径最后一天的价格 axes[1].hist(final_prices, bins15, colorsteelblue, edgecolorwhite, alpha0.8) # 画直方图 axes[1].axvline(xstart_price, colorred, linestyle--, labelf起始价 {start_price}元) # 中图画垂直参考线 axes[1].axvline(xnp.mean(final_prices), colororange, linestyle-, linewidth2, # 中图画垂直参考线 labelf平均终点价 {np.mean(final_prices):.1f}元) # 图例文字 axes[1].set_title(1年后的终点价格分布, fontsize13) # 设置下图标题 axes[1].set_xlabel(最终价格 (元)) # 设置下图横轴日期 axes[1].set_ylabel(股票数量) # 设置下图纵轴 axes[1].legend() # 显示下图图例 axes[1].grid(True, alpha0.3) # 下图显示网格 # --- 右图验证「波动 ∝ 时间的平方根」--- time_points [5, 10, 20, 50, 100, 150, 200, 250] # 选取若干观察日 std_at_time [] # 存放每个时点的价格标准差 for t in time_points: # 代码块开始 prices_at_t [path[t-1] for path in all_paths] # 所有股票在第 t 天的价格 std_at_time.append(np.std(prices_at_t)) # 算标准差 sqrt_time np.sqrt(time_points) # 时间开平方 scale std_at_time[0] / sqrt_time[0] # 缩放系数让理论线对齐第一个点 axes[2].scatter(time_points, std_at_time, s80, colorsteelblue, zorder5, label实际波动(标准差)) # 右图散点 axes[2].plot(time_points, scale * sqrt_time, r--, linewidth2, label理论值: σ ∝ √t) # 右图理论曲线 axes[2].set_title(波动率 vs 时间 (验证√t法则), fontsize13) # 设置右图标题 axes[2].set_xlabel(交易天数) # 执行本行代码 axes[2].set_ylabel(价格标准差 (元)) # 执行本行代码 axes[2].legend() # 执行本行代码 axes[2].grid(True, alpha0.3) # 透明度 plt.tight_layout() # 自动调整子图间距避免标签被裁切 plt.show() # 在 Notebook 里显示图片 # 打印文字结论 print( * 60) # 打印输出 print(实验结论) # 打印输出 print(f 50只股票起始价均为 {start_price}元) # 打印价格统计 print(f 1年后最高价: {max(final_prices):.2f}元) # 打印价格统计 print(f 1年后最低价: {min(final_prices):.2f}元) # 打印价格统计 print(f 1年后平均价: {np.mean(final_prices):.2f}元) # 格式化打印 print(f 价格标准差: {np.std(final_prices):.2f}元) # 格式化打印 print(- * 60) # 打印输出 print( → 每条路径完全不可预测左图) # 打印分隔线或结论 print( → 但大量路径的终点价格呈正态分布中图) # 打印分隔线或结论 print( → 波动幅度与时间平方根成正比右图——Regnault 1860年代的发现) # 打印分隔线或结论 print( * 60) # 打印输出这个实验告诉我们什么这正是巴舍利耶在1900年揭示的核心洞察单只股票的走势无法预测——每条路径都像醉汉走路随机且不重复但大量股票的统计规律是确定的——终点价格服从正态分布波动与√t成正比量化金融的核心逻辑就藏在这里不预测单一结果而是利用统计规律也就是为什么大家常常在讨论大盘 因为如果版块整体走势不好那么个股也极难走出好的曲线。入门实战这里使用最基本最简单的投资策略 即当价格高于5日和20日均线时就买入 由跌穿5日均线就卖出京东方 A 双均线策略的数据获取、回测、报告与绘图。 from __future__ import annotations import time from datetime import datetime from typing import Any, Callable, TextIO import baostock as bs import matplotlib.pyplot as plt import numpy as np import pandas as pd import random STOCK_NAME 京东方A STOCK_CODE 000725 START_DATE 20251201 MAX_RETRIES 5 RETRY_DELAY_SECONDS 5.0 INITIAL_CASH 100000.0 # 初始资金元用于资金账户模拟与总资产曲线 TRADE_COLUMNS [ EntryDate, EntryPrice, ExitDate, ExitPrice, HoldingDays, Return, ] # macOS 常见中文字体优先并保留 Matplotlib 自带字体作为最终回退。 plt.rcParams[font.sans-serif] [ PingFang SC, Arial Unicode MS, Heiti TC, SimHei, DejaVu Sans, ] plt.rcParams[axes.unicode_minus] False def _to_baostock_symbol(code: str) - str: 将 6 位 A 股代码转为 baostock 格式sh.600xxx / sz.000xxx。 baostock 的标的代码需带交易所前缀小写例如京东方A 的 000725 → sz.000725。沪市 6/68 开头、深市 0/30 开头。 code code.strip() if not (code.isdigit() and len(code) 6): raise ValueError(f股票代码格式不正确{code}需为 6 位数字) if code.startswith((60, 68, 11, 13)): return fsh.{code} return fsz.{code} def _query_baostock_daily( symbol: str, start_date: str, end_date: str, ) - pd.DataFrame: 调用 baostock 拉取前复权日线返回原始 DataFrame。 adjustflag2 对应前复权akshare 的 adjustqfq。 baostock 返回的字段值均为字符串数值转换交给调用方处理。 bs.login() try: rs bs.query_history_k_data_plus( symbol, date,code,open,high,low,close,volume,amount,turn, start_datestart_date, end_dateend_date, frequencyd, adjustflag2, # 2前复权3不复权1后复权 ) if rs.error_code ! 0: raise RuntimeError(fbaostock 查询失败{rs.error_code} {rs.error_msg}) rows: list[list[str]] [] while rs.next(): rows.append(rs.get_row_data()) return pd.DataFrame(rows, columnsrs.fields) finally: bs.logout() def fetch_stock_data( sleep_func: Callable[[float], Any] time.sleep, max_retries: int MAX_RETRIES, ) - pd.DataFrame: 获取并标准化京东方 A 前复权日线数据。 数据源为 baostock证券宝免费、无需 token、无反爬限流返回 完整逐日 OHLCV。akshare 的 stock_zh_a_hist 走东方财富接口 常被反爬拦截ConnectionError: Remote end closed connection 故改用 baostock 提升稳定性。 if max_retries 1: raise ValueError(max_retries 必须至少为 1) # baostock 日期格式为 YYYY-MM-DD而 START_DATE 是 YYYYMMDD start_date f{START_DATE[:4]}-{START_DATE[4:6]}-{START_DATE[6:8]} end_date datetime.now().strftime(%Y-%m-%d) bs_symbol _to_baostock_symbol(STOCK_CODE) raw: pd.DataFrame | None None attempt 0 for attempt in range(1, max_retries 1): try: # start_date 2023-01-01 # end_date 2025-01-01 raw _query_baostock_daily( symbolbs_symbol, start_datestart_date, end_dateend_date, ) break except Exception as exc: print(f第 {attempt} 次获取失败{type(exc).__name__}) if attempt max_retries: print(达到最大重试次数请检查网络或稍后再试) raise jitter random.uniform(0.1, 0.9) print(f约 {RETRY_DELAY_SECONDS jitter:g} 秒后重试……) sleep_func(RETRY_DELAY_SECONDS jitter) if raw is None: # pragma: no cover - 循环要么返回数据要么抛出异常 raise RuntimeError(未能获取行情数据) required {date, close, volume} missing required.difference(raw.columns) if missing: raise ValueError(f行情数据缺少必要列{, .join(sorted(missing))}) data raw.rename( columns{ date: Date, open: Open, high: High, low: Low, close: Close, volume: Volume, amount: Amount, turn: Turn, } ).copy() if code in data.columns: data data.drop(columns[code]) # baostock 返回的字段均为字符串统一转成数值 for col in (Open, High, Low, Close, Volume, Amount, Turn): if col in data.columns: data[col] pd.to_numeric(data[col], errorscoerce) try: data[Date] pd.to_datetime(data[Date], errorsraise) except (TypeError, ValueError) as exc: raise ValueError(行情数据的日期列包含无法解析的日期) from exc if data[Date].isna().any(): raise ValueError(行情数据的日期列包含缺失日期) data data.set_index(Date) if data.index.has_duplicates: raise ValueError(行情数据包含重复日期) data data.sort_index() if data.empty: print(未返回行情数据) else: print(f数据获取成功第 {attempt} 次尝试) return data def _empty_trades() - pd.DataFrame: return pd.DataFrame(columnsTRADE_COLUMNS) def backtest_ma_strategy( data: pd.DataFrame, initial_cash: float INITIAL_CASH, ) - tuple[pd.DataFrame, pd.DataFrame]: 按收盘价回测 MA5/MA20 过滤入场、MA5 下穿离场策略。 Position 表示当日收盘后的仓位因此策略收益使用前一日仓位。 initial_cash 为资金账户模拟的初始资金元用于跟踪 现金 / 持仓市值 / 总资产三条曲线。交易按 A 股 100 股一手取整 买入用全部可用现金买整手、卖出清仓。 if not isinstance(data, pd.DataFrame): raise ValueError(data 必须是 pandas DataFrame) if data.empty: raise ValueError(输入数据不能为空) if not isinstance(initial_cash, (int, float)) or initial_cash 0: raise ValueError(initial_cash 必须是大于 0 的正数) if Close not in data.columns: raise ValueError(输入数据缺少 Close 列) if data.index.isna().any(): raise ValueError(输入数据的日期索引包含缺失值) if data.index.has_duplicates: raise ValueError(输入数据的日期索引包含重复日期) result data.copy(deepTrue).sort_index() numeric_close pd.to_numeric(result[Close], errorscoerce) if numeric_close.isna().any(): raise ValueError(Close 列包含缺失或无效的数值) close_values numeric_close.to_numpy(dtypefloat) if not np.isfinite(close_values).all() or (close_values 0).any(): raise ValueError(Close 列必须全部是有限且严格大于 0 的正数不得包含无效值) result[Close] numeric_close.astype(float) result[MA5] result[Close].rolling(window5, min_periods5).mean() result[MA20] result[Close].rolling(window20, min_periods20).mean() result[DailyReturn] result[Close].pct_change(fill_methodNone).fillna(0.0) result[BuySignal] False result[SellSignal] False result[Position] 0 trades: list[dict[str, Any]] [] holding False entry_index: Any None entry_price 0.0 entry_row -1 last_row len(result) - 1 close_col result.columns.get_loc(Close) ma5_col result.columns.get_loc(MA5) ma20_col result.columns.get_loc(MA20) buy_col result.columns.get_loc(BuySignal) sell_col result.columns.get_loc(SellSignal) position_col result.columns.get_loc(Position) for row in range(len(result)): close float(result.iloc[row, close_col]) ma5 result.iloc[row, ma5_col] ma20 result.iloc[row, ma20_col] is_last_row row last_row if holding: crossed_below_ma5 False if row 0 and pd.notna(ma5): previous_close float(result.iloc[row - 1, close_col]) previous_ma5 result.iloc[row - 1, ma5_col] crossed_below_ma5 bool( pd.notna(previous_ma5) and previous_close float(previous_ma5) and close float(ma5) ) if crossed_below_ma5 or is_last_row: result.iloc[row, sell_col] True trades.append( { EntryDate: entry_index, EntryPrice: entry_price, ExitDate: result.index[row], ExitPrice: close, HoldingDays: row - entry_row, Return: close / entry_price - 1.0, } ) holding False else: result.iloc[row, position_col] 1 continue can_enter ( not is_last_row and pd.notna(ma5) and pd.notna(ma20) and close float(ma5) and close float(ma20) ) if can_enter: result.iloc[row, buy_col] True result.iloc[row, position_col] 1 holding True entry_index result.index[row] entry_price close entry_row row result[Position] result[Position].astype(int) result[StrategyReturn] ( result[Position].shift(1).fillna(0) * result[DailyReturn] ) result[StrategyNAV] (1.0 result[StrategyReturn]).cumprod() result[BuyHoldNAV] result[Close] / float(result[Close].iloc[0]) # ---- 资金账户模拟满仓进出A 股 100 股一手取整不足一手的零钱留现金 ---- cash float(initial_cash) shares 0 cash_hist: list[float] [] holdings_hist: list[float] [] equity_hist: list[float] [] executions: list[dict[str, Any]] [] for row in range(len(result)): close float(result.iloc[row, close_col]) date result.index[row] # 同日先卖后买本策略不会同日同时出现此处稳妥处理 if bool(result.iloc[row, sell_col]) and shares 0: proceeds shares * close cash proceeds executions.append({ type: 卖出, date: date, price: close, shares: shares, amount: proceeds, cash_after: cash, }) shares 0 if bool(result.iloc[row, buy_col]) and shares 0: # 用全部可用现金买整手不足 100 股的零钱留在 cash lots int(cash // (close * 100)) buy_shares lots * 100 if buy_shares 0: cost buy_shares * close cash - cost shares buy_shares executions.append({ type: 买入, date: date, price: close, shares: buy_shares, amount: cost, cash_after: cash, }) holdings shares * close cash_hist.append(cash) holdings_hist.append(holdings) equity_hist.append(cash holdings) result[Cash] cash_hist result[HoldingsValue] holdings_hist result[TotalEquity] equity_hist result.attrs[executions] executions trade_frame pd.DataFrame(trades, columnsTRADE_COLUMNS) if trades else _empty_trades() return result, trade_frame def summarize_backtest(result: pd.DataFrame, trades: pd.DataFrame) - dict[str, float | int]: 返回报告所需的可测试指标。 if result.empty: raise ValueError(回测结果不能为空) required {StrategyNAV, BuyHoldNAV} missing required.difference(result.columns) if missing: raise ValueError(f回测结果缺少必要列{, .join(sorted(missing))}) strategy_return float(result[StrategyNAV].iloc[-1] - 1.0) buy_hold_return float(result[BuyHoldNAV].iloc[-1] - 1.0) trade_count int(len(trades)) winning_count int((trades[Return] 0).sum()) if trade_count else 0 win_rate winning_count / trade_count if trade_count else 0.0 summary: dict[str, float | int] { strategy_cumulative_return: strategy_return, buy_hold_return: buy_hold_return, excess_return: strategy_return - buy_hold_return, trade_count: trade_count, winning_count: winning_count, win_rate: float(win_rate), } if TotalEquity in result.columns: initial_cash float(result[Cash].iloc[0] result[HoldingsValue].iloc[0]) final_total float(result[TotalEquity].iloc[-1]) summary[initial_cash] initial_cash summary[final_total_equity] final_total summary[final_cash] float(result[Cash].iloc[-1]) summary[final_holdings_value] float(result[HoldingsValue].iloc[-1]) summary[absolute_profit] final_total - initial_cash return summary def _format_date(value: Any) - str: return value.strftime(%Y-%m-%d) if hasattr(value, strftime) else str(value) def print_report( result: pd.DataFrame, trades: pd.DataFrame, file: TextIO | None None, ) - dict[str, float | int]: 以中文打印逐笔交易及策略汇总并返回汇总指标。 summary summarize_backtest(result, trades) output file print(\n 双均线策略回测报告 , fileoutput) if len(result) 20: print(提示历史数据不足20个交易日均线窗口尚未形成。, fileoutput) if trades.empty: print(暂无交易无交易记录。, fileoutput) else: print(逐笔交易, fileoutput) for number, trade in trades.reset_index(dropTrue).iterrows(): print( f第 {number 1} 笔 | f买入日期 {_format_date(trade[EntryDate])} | f买入价 ¥{trade[EntryPrice]:.2f} | f卖出日期 {_format_date(trade[ExitDate])} | f卖出价 ¥{trade[ExitPrice]:.2f} | f持有日 {int(trade[HoldingDays])} | f单笔收益 {trade[Return]:.2%}, fileoutput, ) print(----------------------------------------, fileoutput) print(f策略累计收益{summary[strategy_cumulative_return]:.2%}, fileoutput) print(f买入持有收益{summary[buy_hold_return]:.2%}, fileoutput) print(f超额收益{summary[excess_return]:.2%}, fileoutput) print(f交易次数{summary[trade_count]}, fileoutput) print(f盈利次数{summary[winning_count]}, fileoutput) print(f胜率{summary[win_rate]:.2%}, fileoutput) if initial_cash in summary: executions result.attrs.get(executions, []) if executions: print(----------------------------------------, fileoutput) print(资金账户明细满仓买整手不足 100 股的零钱留作现金, fileoutput) header ( f{日期:12}{类型:6}{价格:9}{股数:10} f{成交金额:16}{剩余现金:16} ) print(header, fileoutput) for ex in executions: print( f{_format_date(ex[date]):12}{ex[type]:6} f¥{ex[price]:7.2f}{ex[shares]:10,} f¥{ex[amount]:14,.2f}¥{ex[cash_after]:14,.2f}, fileoutput, ) print(----------------------------------------, fileoutput) print(f初始资金¥{summary[initial_cash]:,.2f}, fileoutput) print(f最终总资产¥{summary[final_total_equity]:,.2f}, fileoutput) print(f 其中 现金¥{summary[final_cash]:,.2f}, fileoutput) print(f 其中 股票市值¥{summary[final_holdings_value]:,.2f}, fileoutput) print(f绝对盈亏¥{summary[absolute_profit]:,.2f}, fileoutput) return summary def plot_backtest( result: pd.DataFrame, show: bool True, ) - tuple[plt.Figure, np.ndarray]: 绘制价格/均线/买卖点和策略净值双面板图。 fig, axes plt.subplots( 3, 1, figsize(13, 11), sharexTrue, gridspec_kw{height_ratios: [2, 1, 1]}, ) price_axis, nav_axis, equity_axis axes price_axis.plot(result.index, result[Close], labelClose, colortab:blue, linewidth1.4) price_axis.plot(result.index, result[MA5], labelMA5, colortab:orange, linewidth1.2) price_axis.plot(result.index, result[MA20], labelMA20, colortab:purple, linewidth1.2) buy_points result[result[BuySignal]] sell_points result[result[SellSignal]] price_axis.scatter( buy_points.index, buy_points[Close], marker^, colorgreen, s80, zorder3, labelBuy, ) price_axis.scatter( sell_points.index, sell_points[Close], markerv, colorred, s80, zorder3, labelSell, ) for timestamp, row in buy_points.iterrows(): price_axis.annotate( fB\n¥{row[Close]:.2f}, xy(timestamp, row[Close]), xytext(0, 10), textcoordsoffset points, hacenter, colorgreen, fontsize9, ) for timestamp, row in sell_points.iterrows(): price_axis.annotate( fS\n¥{row[Close]:.2f}, xy(timestamp, row[Close]), xytext(0, -24), textcoordsoffset points, hacenter, colorred, fontsize9, ) price_axis.set_title(f{STOCK_NAME} ({STOCK_CODE}) 双均线策略) price_axis.set_ylabel(价格人民币) price_axis.grid(True, alpha0.3) price_axis.legend() nav_axis.plot( result.index, result[StrategyNAV], labelStrategyNAV, colortab:red, linewidth1.5, ) nav_axis.plot( result.index, result[BuyHoldNAV], labelBuyHoldNAV, colortab:gray, linewidth1.3, ) nav_axis.set_ylabel(净值) nav_axis.grid(True, alpha0.3) nav_axis.legend() # 第三面板资金账户现金 股票市值 总资产 if TotalEquity in result.columns: equity_axis.fill_between( result.index, 0, result[Cash], colortab:gray, alpha0.30, label现金, ) equity_axis.fill_between( result.index, result[Cash], result[TotalEquity], colortab:orange, alpha0.45, label股票市值, ) equity_axis.plot( result.index, result[TotalEquity], label总资产, colortab:green, linewidth1.6, ) equity_axis.set_ylabel(金额元) equity_axis.set_xlabel(日期) equity_axis.grid(True, alpha0.3) equity_axis.legend(locupper left) else: nav_axis.set_xlabel(日期) fig.tight_layout() if show: plt.show() return fig, axes def main() - None: 获取真实数据运行回测打印报告并显示图表。 data fetch_stock_data() print(f\n已获取 {STOCK_NAME}{STOCK_CODE}真实日线数据) print(f交易日数量{len(data)}) if data.empty: print(未获取到可用于回测的数据。) return print(f数据区间{_format_date(data.index[0])} 至 {_format_date(data.index[-1])}) print(f最新收盘价¥{float(data[Close].iloc[-1]):.2f}) result, trades backtest_ma_strategy(data) print_report(result, trades) plot_backtest(result, showTrue) if __name__ __main__: main()可以看出在今年这种极端行情的情况下我们是可以有不错的胜率但是在平常震荡的行情下这套策略就远远不够了注意以上观点不构成投资建议
17天金融量化入门 - Day1
学习目标- [ ] Quant-for-Beginners · 中文零基础量化金融17天速成---课程连接涉嫌广告审核不通过一. 量化入门- 什么是量化金融 √- 你的第一个量化实验- 移动平均线策略- 策略回测二. 风险与组合- 理解波动率- 夏普比率与 Beta- 最大回撤与仓位管理- 多标的组合与相关性---学习产出problematic过去可能需要对代码基础和金融知识有一定的要求而在ai大幅发展的今天这个gap已经大幅度缩小至少在入门阶段就是如此所以这门课应运而生通常来说量化分为下面几个步骤1. 获取数据2. 分析数据3. 设计策略4. 回测策略5. 模拟交易而上面这些步骤都是可以借助ai完成 利用vibe coding生成获取数据和模拟交易的代码利用大模型分析数据和设计策略量化不是预测而是分析是先 计划你的交易 然后再 交易你的计划有的股票热门连续跌了一周有的人就会想 现在已经是谷底是我抄底的好时机 也有的股票连续涨了5天 很多人会因此追涨进去。而量化金融做的是对过去的总结而不对未来的预测。 是让你每一步的决策都有数据理论的支撑而不是一拍脑袋就决定。可视化讲解需要一定的python基础可利用ai辅助理解利用python随即生成一段股票走势图# 导入库 import numpy as np # 数值计算数组、随机数、统计 import pandas as pd # 表格数据处理像 Excel import matplotlib.pyplot as plt # 绘图库折线图、柱状图等 plt.rcParams[font.sans-serif] [SimHei] # 图表中文 plt.rcParams[axes.unicode_minus] False # 坐标轴负号正常显示 np.random.seed(42) # 固定随机数结果可复现 days 120 # 一共模拟 120 个交易日 daily_returns np.random.normal(loc0.0008, scale0.02, sizedays) # 每天随机收益率 price 100 * np.cumprod(1 daily_returns) # 从100元出发连乘成价格序列 df pd.DataFrame({ # 构建表格 收盘价: price, # 原始模拟收盘价 5日均线: pd.Series(price).rolling(5).mean(), # 最近5天均价 20日均线: pd.Series(price).rolling(20).mean() # 最近20天均价 }) # 执行本行代码 plt.figure(figsize(12, 5)) # 创建画布宽12高5英寸 plt.plot(df[收盘价], label收盘价, alpha0.6, linewidth1) # 画折线图 plt.plot(df[5日均线], label5日均线 (MA5), linewidth1.5) # 画折线图 plt.plot(df[20日均线], label20日均线 (MA20), linewidth1.5) # 画折线图 plt.fill_between(range(days), df[5日均线], df[20日均线], alpha0.1, colororange) # 两线之间浅色填充 plt.title(Python 模拟股价走势 移动平均线, fontsize14) # 设置图标题 plt.xlabel(交易日) # 横轴说明 plt.ylabel(价格) # 设置纵轴标签 plt.legend() # 图例 plt.grid(True, alpha0.3) # 浅色网格方便读数 plt.tight_layout() # 自动调整子图间距避免标签被裁切 plt.show() # 在 Notebook 里显示图片 print(f起始价格: ¥{price[0]:.2f}) # 第一天价格 print(f最终价格: ¥{price[-1]:.2f}) # 最后一天价格 print(f区间收益率: {(price[-1]/price[0] - 1)*100:.2f}%) # 整段涨跌幅 print(f最大回撤价格: ¥{price.min():.2f}) # 期间最低价 print(f最高价格: ¥{price.max():.2f}) # 期间最高价布朗运动与随机游走直观感受一下20支股票随机变化曲线这里以8%的增长作为前提这里的中位数是108.14可以看到假如你买了一支股票之后什么也不再继续操作一年后有50%概率是处于中位数之上的也就是年化8%。 问题是现实中股票市场不是8%的增长率同时你也很难找到合适的标的物。利用波动率做风险管理# 实验一布朗运动 / 随机游走 import numpy as np # 数值计算数组、随机数、统计 import matplotlib.pyplot as plt # 绘图库折线图、柱状图等 plt.rcParams[font.sans-serif] [SimHei] # 设置中文字体 plt.rcParams[axes.unicode_minus] False # 坐标轴负号正常显示 np.random.seed(2026) # 固定随机种子 n_stocks 50 # 模拟 50 只「虚拟股票」 n_days 250 # 每只走 250 个交易日约一年 start_price 100 # 起始价都是 100 元 daily_volatility 0.02 # 日波动强度标准差约2% fig, axes plt.subplots(1, 3, figsize(18, 5)) # 1行3列三个子图 # --- 左图画出 50 条价格路径 --- all_paths [] # 用来存每只股票的路径 for _ in range(n_stocks): # 循环 50 次每只股一条路径 daily_returns np.random.normal(0, daily_volatility, n_days) # 每天随机收益 price_path start_price * np.cumprod(1 daily_returns) # 价格连乘 all_paths.append(price_path) # 存入列表 axes[0].plot(price_path, alpha0.3, linewidth0.8) # 画在左图半透明 axes[0].axhline(ystart_price, colorred, linestyle--, alpha0.5, label起始价 100元) # 参考线 axes[0].set_title(50只虚拟股票的随机游走路径, fontsize13) # 设置上图标题 axes[0].set_xlabel(交易日) # 执行本行代码 axes[0].set_ylabel(价格 (元)) # 设置上图纵轴 axes[0].legend() # 显示上图图例 axes[0].grid(True, alpha0.3) # 上图显示网格 # --- 中图一年后的终点价格分布直方图--- final_prices [path[-1] for path in all_paths] # 取每条路径最后一天的价格 axes[1].hist(final_prices, bins15, colorsteelblue, edgecolorwhite, alpha0.8) # 画直方图 axes[1].axvline(xstart_price, colorred, linestyle--, labelf起始价 {start_price}元) # 中图画垂直参考线 axes[1].axvline(xnp.mean(final_prices), colororange, linestyle-, linewidth2, # 中图画垂直参考线 labelf平均终点价 {np.mean(final_prices):.1f}元) # 图例文字 axes[1].set_title(1年后的终点价格分布, fontsize13) # 设置下图标题 axes[1].set_xlabel(最终价格 (元)) # 设置下图横轴日期 axes[1].set_ylabel(股票数量) # 设置下图纵轴 axes[1].legend() # 显示下图图例 axes[1].grid(True, alpha0.3) # 下图显示网格 # --- 右图验证「波动 ∝ 时间的平方根」--- time_points [5, 10, 20, 50, 100, 150, 200, 250] # 选取若干观察日 std_at_time [] # 存放每个时点的价格标准差 for t in time_points: # 代码块开始 prices_at_t [path[t-1] for path in all_paths] # 所有股票在第 t 天的价格 std_at_time.append(np.std(prices_at_t)) # 算标准差 sqrt_time np.sqrt(time_points) # 时间开平方 scale std_at_time[0] / sqrt_time[0] # 缩放系数让理论线对齐第一个点 axes[2].scatter(time_points, std_at_time, s80, colorsteelblue, zorder5, label实际波动(标准差)) # 右图散点 axes[2].plot(time_points, scale * sqrt_time, r--, linewidth2, label理论值: σ ∝ √t) # 右图理论曲线 axes[2].set_title(波动率 vs 时间 (验证√t法则), fontsize13) # 设置右图标题 axes[2].set_xlabel(交易天数) # 执行本行代码 axes[2].set_ylabel(价格标准差 (元)) # 执行本行代码 axes[2].legend() # 执行本行代码 axes[2].grid(True, alpha0.3) # 透明度 plt.tight_layout() # 自动调整子图间距避免标签被裁切 plt.show() # 在 Notebook 里显示图片 # 打印文字结论 print( * 60) # 打印输出 print(实验结论) # 打印输出 print(f 50只股票起始价均为 {start_price}元) # 打印价格统计 print(f 1年后最高价: {max(final_prices):.2f}元) # 打印价格统计 print(f 1年后最低价: {min(final_prices):.2f}元) # 打印价格统计 print(f 1年后平均价: {np.mean(final_prices):.2f}元) # 格式化打印 print(f 价格标准差: {np.std(final_prices):.2f}元) # 格式化打印 print(- * 60) # 打印输出 print( → 每条路径完全不可预测左图) # 打印分隔线或结论 print( → 但大量路径的终点价格呈正态分布中图) # 打印分隔线或结论 print( → 波动幅度与时间平方根成正比右图——Regnault 1860年代的发现) # 打印分隔线或结论 print( * 60) # 打印输出这个实验告诉我们什么这正是巴舍利耶在1900年揭示的核心洞察单只股票的走势无法预测——每条路径都像醉汉走路随机且不重复但大量股票的统计规律是确定的——终点价格服从正态分布波动与√t成正比量化金融的核心逻辑就藏在这里不预测单一结果而是利用统计规律也就是为什么大家常常在讨论大盘 因为如果版块整体走势不好那么个股也极难走出好的曲线。入门实战这里使用最基本最简单的投资策略 即当价格高于5日和20日均线时就买入 由跌穿5日均线就卖出京东方 A 双均线策略的数据获取、回测、报告与绘图。 from __future__ import annotations import time from datetime import datetime from typing import Any, Callable, TextIO import baostock as bs import matplotlib.pyplot as plt import numpy as np import pandas as pd import random STOCK_NAME 京东方A STOCK_CODE 000725 START_DATE 20251201 MAX_RETRIES 5 RETRY_DELAY_SECONDS 5.0 INITIAL_CASH 100000.0 # 初始资金元用于资金账户模拟与总资产曲线 TRADE_COLUMNS [ EntryDate, EntryPrice, ExitDate, ExitPrice, HoldingDays, Return, ] # macOS 常见中文字体优先并保留 Matplotlib 自带字体作为最终回退。 plt.rcParams[font.sans-serif] [ PingFang SC, Arial Unicode MS, Heiti TC, SimHei, DejaVu Sans, ] plt.rcParams[axes.unicode_minus] False def _to_baostock_symbol(code: str) - str: 将 6 位 A 股代码转为 baostock 格式sh.600xxx / sz.000xxx。 baostock 的标的代码需带交易所前缀小写例如京东方A 的 000725 → sz.000725。沪市 6/68 开头、深市 0/30 开头。 code code.strip() if not (code.isdigit() and len(code) 6): raise ValueError(f股票代码格式不正确{code}需为 6 位数字) if code.startswith((60, 68, 11, 13)): return fsh.{code} return fsz.{code} def _query_baostock_daily( symbol: str, start_date: str, end_date: str, ) - pd.DataFrame: 调用 baostock 拉取前复权日线返回原始 DataFrame。 adjustflag2 对应前复权akshare 的 adjustqfq。 baostock 返回的字段值均为字符串数值转换交给调用方处理。 bs.login() try: rs bs.query_history_k_data_plus( symbol, date,code,open,high,low,close,volume,amount,turn, start_datestart_date, end_dateend_date, frequencyd, adjustflag2, # 2前复权3不复权1后复权 ) if rs.error_code ! 0: raise RuntimeError(fbaostock 查询失败{rs.error_code} {rs.error_msg}) rows: list[list[str]] [] while rs.next(): rows.append(rs.get_row_data()) return pd.DataFrame(rows, columnsrs.fields) finally: bs.logout() def fetch_stock_data( sleep_func: Callable[[float], Any] time.sleep, max_retries: int MAX_RETRIES, ) - pd.DataFrame: 获取并标准化京东方 A 前复权日线数据。 数据源为 baostock证券宝免费、无需 token、无反爬限流返回 完整逐日 OHLCV。akshare 的 stock_zh_a_hist 走东方财富接口 常被反爬拦截ConnectionError: Remote end closed connection 故改用 baostock 提升稳定性。 if max_retries 1: raise ValueError(max_retries 必须至少为 1) # baostock 日期格式为 YYYY-MM-DD而 START_DATE 是 YYYYMMDD start_date f{START_DATE[:4]}-{START_DATE[4:6]}-{START_DATE[6:8]} end_date datetime.now().strftime(%Y-%m-%d) bs_symbol _to_baostock_symbol(STOCK_CODE) raw: pd.DataFrame | None None attempt 0 for attempt in range(1, max_retries 1): try: # start_date 2023-01-01 # end_date 2025-01-01 raw _query_baostock_daily( symbolbs_symbol, start_datestart_date, end_dateend_date, ) break except Exception as exc: print(f第 {attempt} 次获取失败{type(exc).__name__}) if attempt max_retries: print(达到最大重试次数请检查网络或稍后再试) raise jitter random.uniform(0.1, 0.9) print(f约 {RETRY_DELAY_SECONDS jitter:g} 秒后重试……) sleep_func(RETRY_DELAY_SECONDS jitter) if raw is None: # pragma: no cover - 循环要么返回数据要么抛出异常 raise RuntimeError(未能获取行情数据) required {date, close, volume} missing required.difference(raw.columns) if missing: raise ValueError(f行情数据缺少必要列{, .join(sorted(missing))}) data raw.rename( columns{ date: Date, open: Open, high: High, low: Low, close: Close, volume: Volume, amount: Amount, turn: Turn, } ).copy() if code in data.columns: data data.drop(columns[code]) # baostock 返回的字段均为字符串统一转成数值 for col in (Open, High, Low, Close, Volume, Amount, Turn): if col in data.columns: data[col] pd.to_numeric(data[col], errorscoerce) try: data[Date] pd.to_datetime(data[Date], errorsraise) except (TypeError, ValueError) as exc: raise ValueError(行情数据的日期列包含无法解析的日期) from exc if data[Date].isna().any(): raise ValueError(行情数据的日期列包含缺失日期) data data.set_index(Date) if data.index.has_duplicates: raise ValueError(行情数据包含重复日期) data data.sort_index() if data.empty: print(未返回行情数据) else: print(f数据获取成功第 {attempt} 次尝试) return data def _empty_trades() - pd.DataFrame: return pd.DataFrame(columnsTRADE_COLUMNS) def backtest_ma_strategy( data: pd.DataFrame, initial_cash: float INITIAL_CASH, ) - tuple[pd.DataFrame, pd.DataFrame]: 按收盘价回测 MA5/MA20 过滤入场、MA5 下穿离场策略。 Position 表示当日收盘后的仓位因此策略收益使用前一日仓位。 initial_cash 为资金账户模拟的初始资金元用于跟踪 现金 / 持仓市值 / 总资产三条曲线。交易按 A 股 100 股一手取整 买入用全部可用现金买整手、卖出清仓。 if not isinstance(data, pd.DataFrame): raise ValueError(data 必须是 pandas DataFrame) if data.empty: raise ValueError(输入数据不能为空) if not isinstance(initial_cash, (int, float)) or initial_cash 0: raise ValueError(initial_cash 必须是大于 0 的正数) if Close not in data.columns: raise ValueError(输入数据缺少 Close 列) if data.index.isna().any(): raise ValueError(输入数据的日期索引包含缺失值) if data.index.has_duplicates: raise ValueError(输入数据的日期索引包含重复日期) result data.copy(deepTrue).sort_index() numeric_close pd.to_numeric(result[Close], errorscoerce) if numeric_close.isna().any(): raise ValueError(Close 列包含缺失或无效的数值) close_values numeric_close.to_numpy(dtypefloat) if not np.isfinite(close_values).all() or (close_values 0).any(): raise ValueError(Close 列必须全部是有限且严格大于 0 的正数不得包含无效值) result[Close] numeric_close.astype(float) result[MA5] result[Close].rolling(window5, min_periods5).mean() result[MA20] result[Close].rolling(window20, min_periods20).mean() result[DailyReturn] result[Close].pct_change(fill_methodNone).fillna(0.0) result[BuySignal] False result[SellSignal] False result[Position] 0 trades: list[dict[str, Any]] [] holding False entry_index: Any None entry_price 0.0 entry_row -1 last_row len(result) - 1 close_col result.columns.get_loc(Close) ma5_col result.columns.get_loc(MA5) ma20_col result.columns.get_loc(MA20) buy_col result.columns.get_loc(BuySignal) sell_col result.columns.get_loc(SellSignal) position_col result.columns.get_loc(Position) for row in range(len(result)): close float(result.iloc[row, close_col]) ma5 result.iloc[row, ma5_col] ma20 result.iloc[row, ma20_col] is_last_row row last_row if holding: crossed_below_ma5 False if row 0 and pd.notna(ma5): previous_close float(result.iloc[row - 1, close_col]) previous_ma5 result.iloc[row - 1, ma5_col] crossed_below_ma5 bool( pd.notna(previous_ma5) and previous_close float(previous_ma5) and close float(ma5) ) if crossed_below_ma5 or is_last_row: result.iloc[row, sell_col] True trades.append( { EntryDate: entry_index, EntryPrice: entry_price, ExitDate: result.index[row], ExitPrice: close, HoldingDays: row - entry_row, Return: close / entry_price - 1.0, } ) holding False else: result.iloc[row, position_col] 1 continue can_enter ( not is_last_row and pd.notna(ma5) and pd.notna(ma20) and close float(ma5) and close float(ma20) ) if can_enter: result.iloc[row, buy_col] True result.iloc[row, position_col] 1 holding True entry_index result.index[row] entry_price close entry_row row result[Position] result[Position].astype(int) result[StrategyReturn] ( result[Position].shift(1).fillna(0) * result[DailyReturn] ) result[StrategyNAV] (1.0 result[StrategyReturn]).cumprod() result[BuyHoldNAV] result[Close] / float(result[Close].iloc[0]) # ---- 资金账户模拟满仓进出A 股 100 股一手取整不足一手的零钱留现金 ---- cash float(initial_cash) shares 0 cash_hist: list[float] [] holdings_hist: list[float] [] equity_hist: list[float] [] executions: list[dict[str, Any]] [] for row in range(len(result)): close float(result.iloc[row, close_col]) date result.index[row] # 同日先卖后买本策略不会同日同时出现此处稳妥处理 if bool(result.iloc[row, sell_col]) and shares 0: proceeds shares * close cash proceeds executions.append({ type: 卖出, date: date, price: close, shares: shares, amount: proceeds, cash_after: cash, }) shares 0 if bool(result.iloc[row, buy_col]) and shares 0: # 用全部可用现金买整手不足 100 股的零钱留在 cash lots int(cash // (close * 100)) buy_shares lots * 100 if buy_shares 0: cost buy_shares * close cash - cost shares buy_shares executions.append({ type: 买入, date: date, price: close, shares: buy_shares, amount: cost, cash_after: cash, }) holdings shares * close cash_hist.append(cash) holdings_hist.append(holdings) equity_hist.append(cash holdings) result[Cash] cash_hist result[HoldingsValue] holdings_hist result[TotalEquity] equity_hist result.attrs[executions] executions trade_frame pd.DataFrame(trades, columnsTRADE_COLUMNS) if trades else _empty_trades() return result, trade_frame def summarize_backtest(result: pd.DataFrame, trades: pd.DataFrame) - dict[str, float | int]: 返回报告所需的可测试指标。 if result.empty: raise ValueError(回测结果不能为空) required {StrategyNAV, BuyHoldNAV} missing required.difference(result.columns) if missing: raise ValueError(f回测结果缺少必要列{, .join(sorted(missing))}) strategy_return float(result[StrategyNAV].iloc[-1] - 1.0) buy_hold_return float(result[BuyHoldNAV].iloc[-1] - 1.0) trade_count int(len(trades)) winning_count int((trades[Return] 0).sum()) if trade_count else 0 win_rate winning_count / trade_count if trade_count else 0.0 summary: dict[str, float | int] { strategy_cumulative_return: strategy_return, buy_hold_return: buy_hold_return, excess_return: strategy_return - buy_hold_return, trade_count: trade_count, winning_count: winning_count, win_rate: float(win_rate), } if TotalEquity in result.columns: initial_cash float(result[Cash].iloc[0] result[HoldingsValue].iloc[0]) final_total float(result[TotalEquity].iloc[-1]) summary[initial_cash] initial_cash summary[final_total_equity] final_total summary[final_cash] float(result[Cash].iloc[-1]) summary[final_holdings_value] float(result[HoldingsValue].iloc[-1]) summary[absolute_profit] final_total - initial_cash return summary def _format_date(value: Any) - str: return value.strftime(%Y-%m-%d) if hasattr(value, strftime) else str(value) def print_report( result: pd.DataFrame, trades: pd.DataFrame, file: TextIO | None None, ) - dict[str, float | int]: 以中文打印逐笔交易及策略汇总并返回汇总指标。 summary summarize_backtest(result, trades) output file print(\n 双均线策略回测报告 , fileoutput) if len(result) 20: print(提示历史数据不足20个交易日均线窗口尚未形成。, fileoutput) if trades.empty: print(暂无交易无交易记录。, fileoutput) else: print(逐笔交易, fileoutput) for number, trade in trades.reset_index(dropTrue).iterrows(): print( f第 {number 1} 笔 | f买入日期 {_format_date(trade[EntryDate])} | f买入价 ¥{trade[EntryPrice]:.2f} | f卖出日期 {_format_date(trade[ExitDate])} | f卖出价 ¥{trade[ExitPrice]:.2f} | f持有日 {int(trade[HoldingDays])} | f单笔收益 {trade[Return]:.2%}, fileoutput, ) print(----------------------------------------, fileoutput) print(f策略累计收益{summary[strategy_cumulative_return]:.2%}, fileoutput) print(f买入持有收益{summary[buy_hold_return]:.2%}, fileoutput) print(f超额收益{summary[excess_return]:.2%}, fileoutput) print(f交易次数{summary[trade_count]}, fileoutput) print(f盈利次数{summary[winning_count]}, fileoutput) print(f胜率{summary[win_rate]:.2%}, fileoutput) if initial_cash in summary: executions result.attrs.get(executions, []) if executions: print(----------------------------------------, fileoutput) print(资金账户明细满仓买整手不足 100 股的零钱留作现金, fileoutput) header ( f{日期:12}{类型:6}{价格:9}{股数:10} f{成交金额:16}{剩余现金:16} ) print(header, fileoutput) for ex in executions: print( f{_format_date(ex[date]):12}{ex[type]:6} f¥{ex[price]:7.2f}{ex[shares]:10,} f¥{ex[amount]:14,.2f}¥{ex[cash_after]:14,.2f}, fileoutput, ) print(----------------------------------------, fileoutput) print(f初始资金¥{summary[initial_cash]:,.2f}, fileoutput) print(f最终总资产¥{summary[final_total_equity]:,.2f}, fileoutput) print(f 其中 现金¥{summary[final_cash]:,.2f}, fileoutput) print(f 其中 股票市值¥{summary[final_holdings_value]:,.2f}, fileoutput) print(f绝对盈亏¥{summary[absolute_profit]:,.2f}, fileoutput) return summary def plot_backtest( result: pd.DataFrame, show: bool True, ) - tuple[plt.Figure, np.ndarray]: 绘制价格/均线/买卖点和策略净值双面板图。 fig, axes plt.subplots( 3, 1, figsize(13, 11), sharexTrue, gridspec_kw{height_ratios: [2, 1, 1]}, ) price_axis, nav_axis, equity_axis axes price_axis.plot(result.index, result[Close], labelClose, colortab:blue, linewidth1.4) price_axis.plot(result.index, result[MA5], labelMA5, colortab:orange, linewidth1.2) price_axis.plot(result.index, result[MA20], labelMA20, colortab:purple, linewidth1.2) buy_points result[result[BuySignal]] sell_points result[result[SellSignal]] price_axis.scatter( buy_points.index, buy_points[Close], marker^, colorgreen, s80, zorder3, labelBuy, ) price_axis.scatter( sell_points.index, sell_points[Close], markerv, colorred, s80, zorder3, labelSell, ) for timestamp, row in buy_points.iterrows(): price_axis.annotate( fB\n¥{row[Close]:.2f}, xy(timestamp, row[Close]), xytext(0, 10), textcoordsoffset points, hacenter, colorgreen, fontsize9, ) for timestamp, row in sell_points.iterrows(): price_axis.annotate( fS\n¥{row[Close]:.2f}, xy(timestamp, row[Close]), xytext(0, -24), textcoordsoffset points, hacenter, colorred, fontsize9, ) price_axis.set_title(f{STOCK_NAME} ({STOCK_CODE}) 双均线策略) price_axis.set_ylabel(价格人民币) price_axis.grid(True, alpha0.3) price_axis.legend() nav_axis.plot( result.index, result[StrategyNAV], labelStrategyNAV, colortab:red, linewidth1.5, ) nav_axis.plot( result.index, result[BuyHoldNAV], labelBuyHoldNAV, colortab:gray, linewidth1.3, ) nav_axis.set_ylabel(净值) nav_axis.grid(True, alpha0.3) nav_axis.legend() # 第三面板资金账户现金 股票市值 总资产 if TotalEquity in result.columns: equity_axis.fill_between( result.index, 0, result[Cash], colortab:gray, alpha0.30, label现金, ) equity_axis.fill_between( result.index, result[Cash], result[TotalEquity], colortab:orange, alpha0.45, label股票市值, ) equity_axis.plot( result.index, result[TotalEquity], label总资产, colortab:green, linewidth1.6, ) equity_axis.set_ylabel(金额元) equity_axis.set_xlabel(日期) equity_axis.grid(True, alpha0.3) equity_axis.legend(locupper left) else: nav_axis.set_xlabel(日期) fig.tight_layout() if show: plt.show() return fig, axes def main() - None: 获取真实数据运行回测打印报告并显示图表。 data fetch_stock_data() print(f\n已获取 {STOCK_NAME}{STOCK_CODE}真实日线数据) print(f交易日数量{len(data)}) if data.empty: print(未获取到可用于回测的数据。) return print(f数据区间{_format_date(data.index[0])} 至 {_format_date(data.index[-1])}) print(f最新收盘价¥{float(data[Close].iloc[-1]):.2f}) result, trades backtest_ma_strategy(data) print_report(result, trades) plot_backtest(result, showTrue) if __name__ __main__: main()可以看出在今年这种极端行情的情况下我们是可以有不错的胜率但是在平常震荡的行情下这套策略就远远不够了注意以上观点不构成投资建议
