【原】設計一個全市場股票查詢及回測方案（附代碼）

AI量化實驗室 2025-05-25 發布于北京

展開全文

原創內容第897篇，專注智能量化投資、個人成長與財富自由。

設計一個全市場股票查詢及回測方案。

要解決以下問題：

由于股票數量比較大，一般是先查詢范圍，比如滬深300成份股，或者某一個行業。

但考慮到回測，不是某一天，而是整體回測周期的。

然后就是篩選指標，比如pe<15以及流動比率>1，技術指標roc(close,25)>0.08等。

這里需要把股票池量價數據，基本面指標，都加載，然后計算技術指標，之后，才能進行回測。

我的方案是使用polars查詢，加載csv的方式。

deepseek的建議是預計算，如果是系統規定好的，常用的，確實可以，而用戶自定義的，就需要動態計算。

預計算有了好處，就是不必全量加載，回測的時候直接查詢即可。

data/├── meta/               # 元數據│   ├── industry.csv    # 行業分類數據│   └── index_constituent.csv  # 指數成分股├── fundamental/        # 基本面數據│   ├── pe.csv│   ├── current_ratio.csv│   └── ...├── market/             # 量價數據│   ├── 2023/│   │   ├── SH600000.csv│   │   └── SZ000001.csv│   └── ...└── technical/          # 預計算技術指標

# industry.csvcode,industrySH600000,銀行SZ000001,綜合
# pe.csvdate,code,pe20230103,SH600000,12.520230103,SZ000001,20.1
# 量價數據文件（SH600000.csv）date,open,high,low,close,vol20230103,8.5,8.7,8.4,8.6,150000

核心代碼：

import polars as plfrom datetime import datefrom typing import List, Dict
class StockBacktestSystem:    def __init__(self, data_path: str):        self.data_path = data_path        self._load_metadata()
    def _load_metadata(self):        # 加載行業和指數成分股數據        self.industry_df = pl.read_csv(f"{self.data_path}/meta/industry.csv")        self.index_constituent = pl.read_csv(            f"{self.data_path}/meta/index_constituent.csv")
    def query_universe(self,                      start_date: str,                     end_date: str,                     index: str = None,                     industry: str = None) -> pl.DataFrame:        """查詢股票池"""        # 實現指數/行業過濾邏輯        pass
    def load_data(self,                codes: List[str],                start_date: str,                end_date: str) -> pl.LazyFrame:        """懶加載所需數據"""        # 加載量價數據        market_dfs = []        for code in codes:            path = f"{self.data_path}/market/{code[:4]}/{code}.csv"            market_dfs.append(                pl.scan_csv(path)                .with_columns(code=pl.lit(code))            )
        # 合并基本面數據        fundamental_df = pl.scan_csv(f"{self.data_path}/fundamental/pe.csv")
        return pl.concat(market_dfs).join(            fundamental_df, on=["date", "code"], how="left")
    def calculate_technical(self,                          df: pl.LazyFrame,                          window: int = 25) -> pl.LazyFrame:        """計算技術指標"""        return df.with_columns([            ((pl.col("close") - pl.col("close").shift(window))             / pl.col("close").shift(window)        ).alias("roc")    ])
    def backtest(self,               start_date: str,               end_date: str,               filters: Dict[str, str],               universe: List[str] = None) -> dict:        """執行回測"""        # 1. 確定股票池        if not universe:            universe = self.query_universe(start_date, end_date)
        # 2. 加載數據        data = self.load_data(universe, start_date, end_date)
        # 3. 計算技術指標        data = self.calculate_technical(data)
        # 4. 應用過濾條件        filtered = data.filter(            (pl.col("pe") < 15) &            (pl.col("current_ratio") > 1) &            (pl.col("roc") > 0.08)        )
        # 5. 回測邏輯（示例）        return self.run_backtest_logic(filtered.collect())
    def run_backtest_logic(self, df: pl.DataFrame) -> dict:        """回測核心邏輯"""        # 實現持倉計算、收益計算等        pass