一、量化金融
1. 基本概念:通常是指运用包括数学、统计学、计算机科学和金融理论在内的多学科知识和方法,通过建立数学模型和算法对大量历史和实时数据的分析和挖掘来进行金融分析、定价、交易和风险管理的一门具有明显跨学科性质的学科领域。
2. 核心目标:用数据驱动的方式寻找市场中的规律和模式,优化投资决策,提高收益或降低风险。
3. 四步闭环:提出假设 → 量化定义 → 历史验证 → 沉淀规则迭代
4. 量化流程:获取数据(Data) → 分析数据(Analysis) → 设计策略(Strategy) → 回测策略(Backtesting) → 模拟交易(Paper Trading)
二、基础概念
1. 收盘价:股票/指数在当天交易结束时的最后一笔成交价格。
2. 持有:你账户里手里有这只股票,当天的涨跌幅会全部算在你头上。比如今天涨了2%,你的账户就赚2%;今天跌了3%,你的账户就亏3%。
3. 空仓:你账户里手里没有这只股票,全部换成了现金。意味着你当天“离场观望”,不参与市场的涨跌。无论当天市场是暴涨还是暴跌,都跟你没关系,当天的收益为 0。
4. 上证指数(全称:上证综合指数)是由上海证券交易所编制的,反映上海证券交易所内所有上市股票整体涨跌情况的“平均价格指标”。通俗理解为“班级平均分”。
5. 基点:指数诞生那一天的“起跑线”或“坐标原点”。
6. 随机游走:想象你站在原点,每一步等概率向左或向右走一步——你的轨迹就是随机游走。股价也类似:每天的涨跌在短期内近似随机。
7. A股:人民币普通股票。通俗地说,就是由在中国内地注册的公司发行,在境内交易所上市,并以人民币进行交易的股票。对于咱们普通投资者来说,可以简单理解为你平时在新闻里听到的“大A”、“中国股市”。
A股主要在两个交易所上市交易:
- 上海证券交易所(上交所):主板股票代码通常以 600、601、603 开头;科创板以 688 开头。
- 深圳证券交易所(深交所):主板股票代码通常以 000 开头;创业板以 300 开头。
8. 复权方式:
| 复权方式 | 做法 | 价格变化 | 核心用途(选哪个?) |
|---|---|---|---|
| 不复权 | 记录实际成交的标价。 | 分红/送股日会出现巨大的价格缺口(跳空)。 | 只能看当天实际标价,无法用于计算真实收益率和技术指标。 |
| 前复权 | 保持最新价格不变,下调所有历史价格。 | 历史价格会变低(早期价格甚至可能是负数)。 | 最常用! 适合看近期走势、画趋势线、计算均线(MA) 和回测。因为保证了技术指标的连续性。 |
| 后复权 | 保持上市首日价格不变,上调所有后来的价格。 | 当前价格会变得非常高(显示股票的真实累积涨幅)。 | 适合看长期总回报(比如茅台从上市到现在涨了多少倍)。 |
三、Python实验
1. 模拟布朗运动——股价真的是"随机游走"吗?
# ========== 实验一:布朗运动 / 随机游走 ========== 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, alpha=0.3, linewidth=0.8) # 画在左图,半透明 axes[0].axhline(y=start_price, color='red', linestyle='--', alpha=0.5, label='起始价 100元') # 参考线 axes[0].set_title('50只虚拟股票的随机游走路径', fontsize=13) # 设置上图标题 axes[0].set_xlabel('交易日') # 执行本行代码 axes[0].set_ylabel('价格 (元)') # 设置上图纵轴 axes[0].legend() # 显示上图图例 axes[0].grid(True, alpha=0.3) # 上图显示网格 # --- 中图:一年后的终点价格分布(直方图)--- final_prices = [path[-1] for path in all_paths] # 取每条路径最后一天的价格 axes[1].hist(final_prices, bins=15, color='steelblue', edgecolor='white', alpha=0.8) # 画直方图 axes[1].axvline(x=start_price, color='red', linestyle='--', label=f'起始价 {start_price}元') # 中图画垂直参考线 axes[1].axvline(x=np.mean(final_prices), color='orange', linestyle='-', linewidth=2, # 中图画垂直参考线 label=f'平均终点价 {np.mean(final_prices):.1f}元') # 图例文字 axes[1].set_title('1年后的终点价格分布', fontsize=13) # 设置下图标题 axes[1].set_xlabel('最终价格 (元)') # 设置下图横轴(日期) axes[1].set_ylabel('股票数量') # 设置下图纵轴 axes[1].legend() # 显示下图图例 axes[1].grid(True, alpha=0.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, s=80, color='steelblue', zorder=5, label='实际波动(标准差)') # 右图:散点 axes[2].plot(time_points, scale * sqrt_time, 'r--', linewidth=2, label='理论值: σ ∝ √t') # 右图:理论曲线 axes[2].set_title('波动率 vs 时间 (验证√t法则)', fontsize=13) # 设置右图标题 axes[2].set_xlabel('交易天数') # 执行本行代码 axes[2].set_ylabel('价格标准差 (元)') # 执行本行代码 axes[2].legend() # 执行本行代码 axes[2].grid(True, alpha=0.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) # 打印输出2. 模拟索普的"概率优势"——微小的胜率偏移如何变成长期利润?
# ========== 实验二:索普——概率优势 + 凯利仓位 ========== 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_rounds = 1000 # 每位玩家模拟下注 1000 轮 n_simulations = 200 # 每种策略重复 200 次(看分布) initial_capital = 1000 # 初始本金 1000 元 win_prob_no_edge = 0.50 # 玩家A:无优势,胜率 50% win_prob_edge = 0.52 # 玩家B/C:有 2% 概率优势 payout_ratio = 1.0 # 赔率 1:1(赢一倍赌注,输一倍赌注) # 凯利公式: f* = (bp - q) / b → 最优下注占本金比例 kelly_fraction = (payout_ratio * win_prob_edge - (1 - win_prob_edge)) / payout_ratio # 凯利公式最优下注比例 aggressive_fraction = 0.25 # 玩家C:有优势但每次下注 25%(太激进) def simulate_player(win_prob, bet_fraction, n_sims=n_simulations): # 定义模拟下注函数 """模拟多人在 n_rounds 轮里的资金曲线。""" # 字典字段 all_curves = [] # 存放每次实验的资金曲线 for _ in range(n_sims): # 外层:重复很多次实验 capital = initial_capital # 本轮起始资金 curve = [capital] # 记录每轮后的资金 for _ in range(n_rounds): # 内层:一轮轮下注 bet = capital * bet_fraction # 本轮下注额 = 本金 × 比例 if np.random.random() < win_prob: # 随机数小于胜率 → 赢 capital += bet * payout_ratio # 赢:拿回赌注并赚一倍 else: # 否则 → 输 capital -= bet # 输:输掉本轮赌注 capital = max(capital, 0.01) # 防止资金变成负数 curve.append(capital) # 执行本行代码 all_curves.append(curve) # 执行本行代码 return np.array(all_curves) # 转成二维数组:行=实验,列=轮次 curves_A = simulate_player(win_prob_no_edge, kelly_fraction) # 无优势 + 凯利 curves_B = simulate_player(win_prob_edge, kelly_fraction) # 有优势 + 凯利 curves_C = simulate_player(win_prob_edge, aggressive_fraction) # 有优势 + 重仓 fig, axes = plt.subplots(1, 3, figsize=(18, 5.5)) # 三个玩家各一张子图 configs = [ # 三个子图的配置 (curves_A, '玩家A: 无优势 (胜率50% + 凯利仓位)', 'red', f'胜率50%, 仓位{kelly_fraction*100:.1f}%'), # 执行本行代码 (curves_B, '玩家B: 索普策略 (胜率52% + 凯利仓位)', 'green', f'胜率52%, 仓位{kelly_fraction*100:.1f}%'), # 执行本行代码 (curves_C, '玩家C: 有优势但冒进 (胜率52% + 重仓)', 'goldenrod', f'胜率52%, 仓位25%'), # 执行本行代码 ] # 数组拼接结束 for ax, (curves, title, color, label) in zip(axes, configs): # 为每个玩家画子图 for i in range(min(30, n_simulations)): # 只画前30条细线,避免太乱 ax.plot(curves[i], alpha=0.15, linewidth=0.6, color=color) # 在子图上画折线 median_curve = np.median(curves, axis=0) # 200次实验的中位数轨迹 ax.plot(median_curve, color=color, linewidth=2.5, label=f'中位数轨迹 ({label})') # 在子图上画折线 ax.axhline(y=initial_capital, color='gray', linestyle='--', alpha=0.5, label=f'初始本金 {initial_capital}元') # 画水平参考线 ax.set_title(title, fontsize=12, fontweight='bold') # 设置子图标题 ax.set_xlabel('下注轮次') # 设置子图横轴 ax.set_ylabel('资金 (元)') # 设置子图纵轴 ax.legend(fontsize=9, loc='upper left') # 显示图例 ax.grid(True, alpha=0.3) # 显示网格 ax.set_yscale('log') # 纵轴用对数刻度,差距大时更好看 ax.set_ylim(1, None) # 设置纵轴范围 plt.tight_layout() # 自动调整子图间距,避免标签被裁切 plt.show() # 在 Notebook 里显示图片 print("=" * 70) # 打印输出 print(f"凯利公式计算: 最优下注比例 f* = {kelly_fraction*100:.2f}% (胜率52%, 赔率1:1)") # 打印凯利公式结果 print("=" * 70) # 打印输出 for name, curves, color in [("玩家A (无优势)", curves_A, "red"), # 打印各玩家统计 ("玩家B (索普策略)", curves_B, "green"), # 执行本行代码 ("玩家C (有优势但冒进)", curves_C, "goldenrod")]: # 代码块开始 finals = curves[:, -1] # 每个实验最后一轮的资金 win_rate = np.mean(finals > initial_capital) * 100 # 最终赚钱的比例 median_final = np.median(finals) # 赋值:median_final print(f"\n 【{name}】") # 格式化打印 print(f" 中位数终点资金: {median_final:,.0f}元") # 格式化打印 print(f" 盈利概率: {win_rate:.1f}%") # 格式化打印 print(f" 最好情况: {np.max(finals):,.0f}元 | 最差情况: {np.min(finals):,.2f}元") # 格式化打印 print("\n" + "=" * 70) # 打印输出 print(" → 没有概率优势,凯利公式也救不了你(玩家A)") # 打印分隔线或结论 print(" → 仅仅2%的胜率优势 + 科学仓位管理 = 长期稳定复利(玩家B)") # 打印分隔线或结论 print(" → 有优势但仓位过重,反而可能亏光(玩家C)") # 打印分隔线或结论 print(" → 这就是索普的核心发现:概率优势 × 仓位管理 = 量化盈利的底层公式") # 打印分隔线或结论 print("=" * 70) # 打印输出3. 本章爽点:你拿到真实股票数据了
# ========== 第一章爽点:下载真实股票数据(baostock 版) ========== import warnings warnings.filterwarnings('ignore') import matplotlib.pyplot as plt import pandas as pd import baostock as bs from datetime import datetime, timedelta # 设置中文字体 plt.rcParams['font.sans-serif'] = ['SimHei'] plt.rcParams['axes.unicode_minus'] = False # ---------- 1. 登录 baostock ---------- lg = bs.login() if lg.error_code != '0': print(f"登录失败: {lg.error_msg}") exit() # ---------- 2. 计算起止日期 ---------- end_date = datetime.today().strftime('%Y-%m-%d') # 格式 'YYYY-MM-DD' start_date = (datetime.today() - timedelta(days=180)).strftime('%Y-%m-%d') # ---------- 3. 下载 A 股历史日线数据(前复权) ---------- # 股票代码:sh.600000 或 sz.000001(示例使用平安银行 sz.000001) rs = bs.query_history_k_data_plus( code="sz.000001", fields="date,open,high,low,close,volume", start_date=start_date, end_date=end_date, frequency="d", adjustflag="2" # 2 表示前复权 ) if rs.error_code != '0': print(f"数据查询失败: {rs.error_msg}") bs.logout() exit() # 提取数据为 DataFrame data_list = [] while (rs.error_code == '0') & rs.next(): data_list.append(rs.get_row_data()) df = pd.DataFrame(data_list, columns=rs.fields) # 检查是否为空 if df.empty: print("未获取到数据,请检查股票代码或日期") bs.logout() exit() # 数据类型转换 df['date'] = pd.to_datetime(df['date']) df['open'] = df['open'].astype(float) df['high'] = df['high'].astype(float) df['low'] = df['low'].astype(float) df['close'] = df['close'].astype(float) df['volume'] = df['volume'].astype(float) # 设置日期为索引,并按时间排序 df.set_index('date', inplace=True) df.sort_index(inplace=True) # 重命名列(与之前习惯保持一致) df.rename(columns={ 'open': 'Open', 'high': 'High', 'low': 'Low', 'close': 'Close', 'volume': 'Volume' }, inplace=True) # ---------- 4. 退出 baostock ---------- bs.logout() # ---------- 5. 输出信息 ---------- print('🎉 恭喜!你已经拿到真实股票数据') print(f' 共 {len(df)} 个交易日') print(f' 最新收盘价: {df["Close"].iloc[-1]:.2f} 元') display(df.tail(5)) # ---------- 6. 绘图 ---------- fig, axes = plt.subplots(2, 1, figsize=(12, 6), sharex=True, gridspec_kw={'height_ratios': [3, 1]}) axes[0].plot(df.index, df['Close'], color='tab:blue', linewidth=1.5) axes[0].set_title('真实数据 · 平安银行 000001 收盘价', fontsize=14) axes[0].set_ylabel('元') axes[0].grid(True, alpha=0.3) axes[1].bar(df.index, df['Volume'], width=0.8, color='gray', alpha=0.5) axes[1].set_ylabel('成交量') axes[1].set_xlabel('日期') axes[1].grid(True, alpha=0.3) plt.tight_layout() plt.show()