1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
|
# 基于 AKShare 的行情数据采集 Pipeline
import akshare as ak
import polars as pl
from pathlib import Path
from datetime import date, timedelta
import logging
logger = logging.getLogger(__name__)
class MarketDataCollector:
"""A股日频行情数据采集器"""
def __init__(self, storage_path: str = "./data/market"):
self.storage_path = Path(storage_path)
self.storage_path.mkdir(parents=True, exist_ok=True)
def fetch_daily_bars(self, trade_date: str) -> pl.DataFrame:
"""获取指定日期全市场日K线数据"""
try:
df = ak.stock_zh_a_spot_em()
# 转换为 Polars DataFrame 以获得更好的性能
result = pl.DataFrame({
"trade_date": pl.Series([trade_date] * len(df)),
"symbol": df["代码"].tolist(),
"name": df["名称"].tolist(),
"open": df["今开"].tolist(),
"high": df["最高"].tolist(),
"low": df["最低"].tolist(),
"close": df["最新价"].tolist(),
"volume": df["成交量"].tolist(),
"amount": df["成交额"].tolist(),
"turnover_rate": df["换手率"].tolist(),
"pe_ratio": df["市盈率-动态"].tolist(),
"pb_ratio": df["市净率"].tolist(),
"total_mv": df["总市值"].tolist(),
"circ_mv": df["流通市值"].tolist(),
})
return result
except Exception as e:
logger.error(f"Failed to fetch daily bars for {trade_date}: {e}")
raise
def save_parquet(self, df: pl.DataFrame, trade_date: str):
"""以Parquet格式分区存储"""
path = self.storage_path / f"date={trade_date}" / "daily_bars.parquet"
path.parent.mkdir(parents=True, exist_ok=True)
df.write_parquet(path, compression="zstd")
def run_daily_pipeline(self, start_date: str, end_date: str):
"""批量采集历史数据"""
dates = pl.date_range(
date.fromisoformat(start_date),
date.fromisoformat(end_date),
interval="1d",
eager=True
)
for d in dates:
date_str = d.strftime("%Y-%m-%d")
logger.info(f"Collecting data for {date_str}")
df = self.fetch_daily_bars(date_str)
self.save_parquet(df, date_str)
|
