A股量化入门实战：基于AKShare的数据获取与策略回测全流程解析

1次阅读

共计 2720 个字符，预计需要花费 7 分钟才能阅读完成。

刚接触 A 股量化的开发者常会遇到这样的场景：好不容易写好了策略代码，却发现历史数据获取要么收费高昂，要么接口不稳定。相比美股成熟的量化生态，A 股市场存在几个典型痛点：

数据孤岛问题：核心数据分散在各券商、交易所、第三方平台
非标数据格式：不同来源的复权方式、字段命名差异显著
更新延迟：免费源常有 15 分钟以上的行情延迟

我曾尝试过多个数据源，直到发现 AKShare 这个宝藏库——一个免费、开源的金融数据接口库，特别适合个人开发者低成本启动。

先看几个主流数据源的特性对比（以股票基础数据为例）：

特性	AKShare	Tushare Pro	Wind	聚源
费用	免费	分级收费	年费制	按量计费
更新频率	实时	15 分钟延迟	实时	实时
历史数据深度	10 年 +	20 年 +	完整	完整
接口稳定性	★★★☆	★★★★	★★★★★	★★★★

AKShare 的核心优势 在于：

完全规避了商业 API 的鉴权复杂性问题
通过 requests_html 直接抓取权威源数据，避免二次加工
支持银行间市场、期货、期权等泛金融数据

安装只需要一行命令（建议使用虚拟环境）：

pip install akshare pandas backtrader

以获取贵州茅台 (600519) 的日线数据为例：

import akshare as ak

# 获取后复权数据
def get_stock_data(stock_code):
    try:
        df = ak.stock_zh_a_daily(
            symbol=stock_code, 
            adjust="hfq"  # hfq 后复权, qfq 前复权, none 不复权
        )
        # 处理停牌日（AKShare 已自动过滤）df = df[df["volume"] > 0]  
        # 标准化字段名以兼容 Backtrader
        df.rename(columns={
            "date": "datetime",
            "open": "open",
            "high": "high",
            "low": "low",
            "close": "close",
            "volume": "volume"
        }, inplace=True)
        df["datetime"] = pd.to_datetime(df["datetime"])
        df.set_index("datetime", inplace=True)
        return df
    except Exception as e:
        print(f"数据获取异常: {e}")
        return None

# 示例调用
mt_data = get_stock_data("sh600519")
print(mt_data.tail(3))

关键点说明：
– 复权处理直接影响策略回测结果，通常用后复权 (hfq) 做长期回测
– AKShare 返回的 DataFrame 需要转换为 Backtrader 要求的格式
– 建议添加重试机制应对网络波动

接下来用 Backtrader 实现经典的双均线策略：

import backtrader as bt

class DualMAStrategy(bt.Strategy):
    params = (('fast_period', 10),
        ('slow_period', 30),
    )

    def __init__(self):
        # 计算两条均线
        self.fast_ma = bt.indicators.SMA(
            self.data.close, 
            period=self.p.fast_period
        )
        self.slow_ma = bt.indicators.SMA(
            self.data.close, 
            period=self.p.slow_period
        )
        self.crossover = bt.indicators.CrossOver(
            self.fast_ma, 
            self.slow_ma
        )

    def next(self):
        if not self.position:  # 没有持仓
            if self.crossover > 0:  # 金叉
                self.buy(size=100)  # 买入 100 股
        elif self.crossover < 0:  # 死叉
            self.close()  # 平仓

# 回测引擎配置
def run_backtest(data):
    cerebro = bt.Cerebro()

    # 加载数据
    data_feed = bt.feeds.PandasData(dataname=data)
    cerebro.adddata(data_feed)

    # 添加策略
    cerebro.addstrategy(DualMAStrategy)

    # 设置初始资金
    cerebro.broker.setcash(100000.0)

    # 运行回测
    print('初始资金: %.2f' % cerebro.broker.getvalue())
    cerebro.run()
    print('最终资金: %.2f' % cerebro.broker.getvalue())

    # 可视化（需要安装 matplotlib）cerebro.plot(style='candlestick')

# 执行回测
run_backtest(mt_data)

实际部署时需要特别注意：

限流应对：
AKShare 默认没有严格限流，但频繁请求可能触发源站反爬

建议添加随机延时：

import random, time
time.sleep(random.uniform(0.5, 1.5))

数据缓存：
使用 pickle 或feather格式本地存储历史数据

实现增量更新逻辑：

if os.path.exists("cache.pkl"):
    old_df = pd.read_pickle("cache.pkl")
    new_df = get_stock_data("sh600519")
    updated_df = pd.concat([old_df, new_df]).drop_duplicates()