量化交易系统之数据模块

介绍

如何获取股票数据困扰了我很久。我总是想着完美主义，导致我屡屡受挫。直到我幡然领悟，够用就行，将我解放出来。

本文不打算构建一套至善至美的大系统，而是从个人角度触发，针对量化回测场景，如何够用就行，如何能省则省。

存储格式

存储目录

首先，指定一个目录，用于存放行情数据。可以采用多层结构：

D:\TradeData：存放所有股票数据
D:\TradeData\stock：专门存放股票行情数据

之所以采用多层结构，主要是方便未来存放其它类型数据。

由于只是自己用，为了图省事，直接硬编码在 Python 代码中。

文件命名规范

股票数据从 SDK 获取到之后，是一个 Pandas Dataframe，通常会存为 csv 格式。

csv 命名规范满足：[symbol_name][start_date][end_date].csv

其中：

start_date 是开始日期，2022-01-01
end_date：是结束日期，2022-01-01
symbol_name：是股票代号

如果你细想的话，会发现这里面存在一些问题：

时间区间会有很多重叠部分，比如 A 文件 2000-01-01~2000-12-31，B 文件 2000-02-01~2000-03-01，其中有段冗余部分，怎么办？不办，重了就重了。
不同 SDK 的 symbol_name 是不同的，有的叫 000001.sh，有的叫 sh000001，怎么办？不怎么办，爱叫啥叫啥，井水不犯河水。

其实，我想做成一个磁盘缓存，给出 Key 本地找，有就直接返回，没有就拉取保存。

代码实现

注意：因为是回测用的，不是看盘用的，所有数据都采用后复权。

一共 70 多行代码搞定：

import akshare as ak
from pathlib import Path
import efinance as ef
import pandas as pd

STOCK_DATA_PATH = Path("D:\\Ray\\RaySystemData\\stock\\")


# 使用 akshare 拉取数据
def get_stock_ak(symbol: str, start: str, end: str) -> pd.DataFrame:
    """
    symbol: "000001"
    start: "19700101"
    end: "19700101"
    """
    STOCK_DATA_PATH.mkdir(exist_ok=True)
    path = STOCK_DATA_PATH.joinpath(_gen_filename(symbol, start, end))
    if _is_exist(path):
        return _read(path)
    else:
        return _save(_download_ak(symbol, start, end), path)


# 使用 efinance 拉取数据
def get_stock_ef(symbol: str, start: str, end: str) -> pd.DataFrame:
    STOCK_DATA_PATH.mkdir(exist_ok=True)
    path = STOCK_DATA_PATH.joinpath(_gen_filename(symbol, start, end))
    if _is_exist(path):
        return _read(path)
    else:
        return _save(_download_ef(symbol, start, end), path)


def _is_exist(path: Path) -> bool:
    return path.exists()


def _read(path: Path) -> pd.DataFrame:
    return pd.read_csv(path)


def _download_ak(symbol: str, start: str, end: str) -> pd.DataFrame:
    data = ak.stock_zh_a_hist(
        symbol=symbol, start_date=start, end_date=end, adjust="hfq", period="daily"
    )
    data.日期 = pd.to_datetime(data.日期)
    data.set_index("日期", drop=False, inplace=True)
    return data


def _download_ef(symbol: str, start: str, end: str) -> pd.DataFrame:
    """
    后复权: 2
    日频数据 101
    """
    data: pd.DataFrame = ef.stock.get_quote_history(  # type: ignore
        symbol, beg=start, end=end, fqt=2, klt=101
    )
    data.日期 = pd.to_datetime(data.日期)
    data.set_index("日期", drop=False, inplace=True)
    return data


def _save(data: pd.DataFrame, path: Path):
    data.to_csv(path)
    return data


def _gen_filename(symbol: str, start: str, end: str):
    return f"[{symbol}][{start}][{end}].csv"


if __name__ == "__main__":
    get_stock_ak("600377", "20000101", "20201231")