原創內容第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.csv code,industry SH600000,銀行 SZ000001,綜合
# pe.csv date,code,pe 20230103,SH600000,12.5 20230103,SZ000001,20.1
# 量價數據文件(SH600000.csv) date,open,high,low,close,vol 20230103,8.5,8.7,8.4,8.6,150000
import polars as pl from datetime import date from 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 但都需要把價量數據,基本面數據,估計數據加載到內存,然后計算擴展指標。長期主義,要求目光長遠,志向遠大,持久地布局、堅持做一件事,拿結果,獲得理想的狀態。而活在當下呢,又因為活在未來的人焦慮啊,當上才是真實的存在。人生在于體驗啊,誰知道,明天和意外哪一個先來呢?再引申,會不會變成“有花堪折真須折,莫待無花空折枝”呢。18歲的時候,再獲得8歲時想要的玩具,顯然早已沒有了興趣。《拿鐵因素》這三本講了一堆故事,其實就講三句話:先支付自己,不要做預算,讓賬戶自動運轉,從現在開始,富足地生活。這本書講普通人如何財富自由的,很好的平衡了上述的關系。考慮長期主義,無論你收入水平如何,你先支付自己,比如把每天收入的1/8存下來,看起來不難吧。然后就不要做預算了——以前,講財富積累的書,總教人記賬,且不論是否堅持得了,關鍵是生活就變成很無趣,沒有了當下,就是如何省,如何不花錢。——只有未來,沒有現在。而作者有兩句話就解決了這個問題。——存下每天收的1/8,其余的,就不必預算。這個1/8進入金錢系統,自動運轉。——這個系統在未來40年,會讓你財富自由,這就構建了一個“時間的朋友”。長期主義的問題就解決了,富足地生活,就可以做你想做的事情。同樣的,如果構建一人企業的“商業系統”,讓它能自動運轉,那么財富積累的速度還會加快。比如,類似的,每天花1/8的時間去構建這個“商業系統”,這個商業系統構建好了,就可以源源不斷帶來現金流。那么,你就提前財富自由了。同樣,不需要做所謂的“時間管理”,什么把時間分成多少段,然后如何管理效率最高之類的,且不論能不能堅持,就算能堅持,那也很無趣啊。簡言之,平衡長期主義與活在當下,就是把1/8的收入和1/8的注意力,首先支付給自己,搭建可以自主運行的“金錢系統”和“商業系統”。然后,就富足的生活吧。(這里1/8僅是一個參考值)。你就不必為了說今天吃了頓大餐,買了個貴重的數碼產品而自責,也不必為了刷了一中午的短視頻而自責。因為,如果這就是你喜歡的當下,為什么不可以呢。——你的長期主義已經在運作了,你已經是時間的朋友!代碼和數據下載:AI量化實驗室——2025量化投資的星辰大海AI量化實驗室 星球,已經運行三年多,1700+會員。 aitrader代碼,因子表達式引擎、遺傳算法(Deap)因子挖掘引擎等,支持vnpy,qlib,backtrader和bt引擎,內置多個年化30%+的策略,每周五迭代一次,代碼和數據在星球全部開源。
點擊 “查看原文”,直接訪問策略集合。
|