通過akshare庫，獲取股票數據，并生成TabPFN這個模型 可以識別、處理的格式，寫一個完整的預處理示例，并構建一個預測未來 3 天股價漲跌的分類任務

用TabPFN這個模型構建一個預測未來 3 天股價漲跌的分類任務，進行預測并輸出準確率

代碼

import os
import sys
import pandas as pd
import akshare as ak
# try:
#     import akshare as ak
# except ImportError:
#     print("請先運行: pip install akshare")
#     sys.exit(1)def fetch_stock_data(stock_code="000001", start_date="20150101", end_date=None):"""獲取股票歷史行情數據，兼容不同akshare接口"""print(f"DEBUG: akshare version: {ak.__version__}")print(f"DEBUG: fetch_stock_data params: symbol={stock_code}, start_date={start_date}, end_date={end_date}")try:# 優先嘗試 stock_zh_a_dailydf = ak.stock_zh_a_daily(symbol=stock_code, adjust="qfq")print("DEBUG: 使用 ak.stock_zh_a_daily 成功")print("DEBUG: Columns before rename:", df.columns.tolist())print("DEBUG: Head before rename:\n", df.head())# 若有 start_date/end_date，篩選if "date" in df.columns:df["date"] = pd.to_datetime(df["date"])if start_date:df = df[df["date"] >= pd.to_datetime(start_date)]if end_date:df = df[df["date"] <= pd.to_datetime(end_date)]df = df.sort_values("date").reset_index(drop=True)except Exception as e:print("ERROR: ak.stock_zh_a_daily() 調用異常，嘗試 fallback 原接口")print(f"Exception: {e}")try:df = ak.stock_zh_a_hist(symbol=stock_code, period="daily", start_date=start_date, end_date=end_date, adjust="qfq")print("DEBUG: 使用 ak.stock_zh_a_hist 成功")print("DEBUG: Columns before rename:", df.columns.tolist())print("DEBUG: Head before rename:\n", df.head())except Exception as e2:print("ERROR: Exception occurred while fetching stock data by both interfaces!")print(f"Exception: {e2}")return pd.DataFrame()if df.empty:print("ERROR: Fetched DataFrame is empty! Check stock code, date range, or network/API issues.")return df# 兼容列名rename_map = {"日期": "date","開盤": "open","收盤": "close","最高": "high","最低": "low","成交量": "volume","成交額": "amount","振幅": "amplitude","漲跌幅": "pct_chg","漲跌額": "chg","換手率": "turnover"}for k in list(rename_map.keys()):if k not in df.columns:rename_map.pop(k)df = df.rename(columns=rename_map)if "date" in df.columns:df["date"] = pd.to_datetime(df["date"])df = df.sort_values("date").reset_index(drop=True)# 自動補充pct_chg列（漲跌幅），百分比格式if "pct_chg" not in df.columns and "close" in df.columns:df["pct_chg"] = df["close"].pct_change() * 100return dfdef create_features_and_labels(df, n_past=5, n_future=3):"""構造特征和標簽，標簽為未來3日漲跌（1=漲，0=跌/平）"""feats = []labels = []for idx in range(n_past, len(df) - n_future):past_slice = df.iloc[idx-n_past:idx]# 特征: 過去n_past日的收盤價、漲跌幅、成交量feature = []feature += list(past_slice["close"].values)feature += list(past_slice["pct_chg"].values)feature += list(past_slice["volume"].values)# 未來n_future日的收盤價均值future_close_mean = df.iloc[idx:idx+n_future]["close"].mean()curr_close = df.iloc[idx-1]["close"]# 漲跌標簽: 未來3日均值 > 當前收盤價 => 1，否則0label = 1 if future_close_mean > curr_close else 0feats.append(feature)labels.append(label)feats_df = pd.DataFrame(feats, columns=[f"close_t-{i}" for i in range(n_past,0,-1)] + [f"pct_chg_t-{i}" for i in range(n_past,0,-1)] + [f"volume_t-{i}" for i in range(n_past,0,-1)])feats_df["label"] = labelsreturn feats_dfdef save_for_tabpfn(df, out_csv):"""保存為TabPFN模型可讀取的csv格式"""df.to_csv(out_csv, index=False)print(f"已保存至: {out_csv}")print(df.head())def main():# 拉取A股代碼表，打印前10條，輔助判斷symbol格式print("嘗試拉取A股代碼表，輔助symbol格式判斷...")try:code_df = ak.stock_info_a_code_name()print("A股代碼表前10條：")print(code_df.head(10))print("平安銀行相關行：")print(code_df[code_df["code"].str.contains("000001")])except Exception as e:print(f"拉取A股代碼表失敗: {e}")stock_code = "sz000001"  # 平安銀行start_date = "20150101"print("正在獲取股票數據...")df = fetch_stock_data(stock_code, start_date)print("正在生成特征與標簽...")processed = create_features_and_labels(df, n_past=5, n_future=3)# out_csv = os.path.join(os.path.dirname(__file__), f"{stock_code}_tabpfn.csv")out_csv =  f"{stock_code}_tabpfn.csv"print("正在保存為TabPFN格式...")save_for_tabpfn(processed, out_csv)print("預處理完成。")if __name__ == "__main__":main()

預測

import os
# 臨時繞過 /dev/null 權限問題
os.devnull = "/tmp/null"
if not os.path.exists("/tmp/null"):with open("/tmp/null", "w") as f:passimport sys
import numpy as np
import pandas as pd# 自動安裝tabpfn（如未安裝）
try:from tabpfn import TabPFNClassifier
except ImportError:import subprocesssubprocess.check_call([sys.executable, "-m", "pip", "install", "tabpfn"])from tabpfn import TabPFNClassifier# 讀取數據
# DATA_PATH = os.path.join(os.path.dirname(__file__), "sz000001_tabpfn.csv")
DATA_PATH = "sz000001_tabpfn.csv"
data = pd.read_csv(DATA_PATH, header=None)
X = data.iloc[:, :-1].values
y = data.iloc[:, -1].values# 按時間順序劃分（前80%訓練，后20%測試）
split_idx = int(0.8 * len(X))
X_train, X_test = X[:split_idx], X[split_idx:]
y_train, y_test = y[:split_idx], y[split_idx:]# TabPFN訓練與預測
# clf = TabPFNClassifier(device='cpu')
clf = TabPFNClassifier(device='cuda')
clf.fit(X_train, y_train)
y_pred = clf.predict(X_test)
accuracy = np.mean(y_pred == y_test)print(f"Test Accuracy: {accuracy:.4f}")

最后生成的準確率:

Test Accuracy: 0.4812

這個準確率低于50%,反而可能證明程序是對的。