feat(cli): sync export — 日线按年导出 Parquet(data/parquet/stock_daily/<year>.parquet)
- 对应 ROADMAP §1.4「历史时序大数据转 Parquet」;pyarrow 随 qlib 依赖已可用 - 用法:uv run python -m app.cli.sync export [--years 2023,2024] - 实测:9680 行 → 2023/2024 两个 parquet(data/parquet 已被 gitignore) - ruff clean / pytest 86 passed
This commit is contained in:
@@ -169,6 +169,55 @@ def cmd_verify(args) -> int:
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_export(args) -> int:
|
||||
"""把 SQLite 日线按年导出为 Parquet(data/parquet/stock_daily/<year>.parquet)。"""
|
||||
from pathlib import Path
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from app.infrastructure.persistence.sqlalchemy.models.market import StockDailyModel
|
||||
|
||||
settings = get_settings()
|
||||
out_root = settings.storage.get("parquet_dir") or Path("data/parquet")
|
||||
out_root.mkdir(parents=True, exist_ok=True)
|
||||
total = 0
|
||||
years = [int(y) for y in (args.years or "").split(",") if y.strip()] or None
|
||||
with _session_ctx() as session:
|
||||
all_bars = session.execute(
|
||||
select(StockDailyModel).order_by(StockDailyModel.trade_date)
|
||||
).scalars()
|
||||
frame = pd.DataFrame(
|
||||
[
|
||||
{
|
||||
"symbol": b.symbol,
|
||||
"trade_date": b.trade_date,
|
||||
"open": float(b.open) if b.open is not None else None,
|
||||
"high": float(b.high) if b.high is not None else None,
|
||||
"low": float(b.low) if b.low is not None else None,
|
||||
"close": float(b.close) if b.close is not None else None,
|
||||
"volume": float(b.volume) if b.volume is not None else None,
|
||||
"amount": float(b.amount) if b.amount is not None else None,
|
||||
}
|
||||
for b in all_bars
|
||||
]
|
||||
)
|
||||
if frame.empty:
|
||||
print("[export] 无日线数据,请先运行 sync daily")
|
||||
return 0
|
||||
frame["trade_date"] = pd.to_datetime(frame["trade_date"])
|
||||
out_dir = out_root / "stock_daily"
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
for year, group in frame.groupby(frame["trade_date"].dt.year):
|
||||
if years and int(year) not in years:
|
||||
continue
|
||||
path = out_dir / f"{year}.parquet"
|
||||
group.sort_values(["symbol", "trade_date"]).to_parquet(path, index=False)
|
||||
total += len(group)
|
||||
print(f"[export] {year} → {path}({len(group)} 行)")
|
||||
print(f"[export] 合计 {total} 行 → {out_dir}")
|
||||
return 0
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(prog="app.cli.sync", description="Tushare 数据同步 CLI")
|
||||
sub = parser.add_subparsers(dest="command", required=True)
|
||||
@@ -197,6 +246,10 @@ def build_parser() -> argparse.ArgumentParser:
|
||||
p_verify = sub.add_parser("verify", help="新浪交叉验证最新行情")
|
||||
p_verify.add_argument("--symbol", required=True)
|
||||
p_verify.set_defaults(func=cmd_verify)
|
||||
|
||||
p_export = sub.add_parser("export", help="日线按年导出 Parquet(data/parquet)")
|
||||
p_export.add_argument("--years", default="", help="逗号分隔年份,留空导出全部")
|
||||
p_export.set_defaults(func=cmd_export)
|
||||
return parser
|
||||
|
||||
|
||||
|
||||
@@ -108,6 +108,7 @@ uv run python -m app.cli.sync daily --symbols 600519.SH,000858.SZ --start 202301
|
||||
uv run python -m app.cli.sync daily --all --start 20240101 --resume # 全市场 + 断点续传
|
||||
uv run python -m app.cli.sync financial --symbols 600519.SH
|
||||
uv run python -m app.cli.sync verify --symbol 600519.SH # 新浪交叉验证
|
||||
uv run python -m app.cli.sync export --years 2024 # 日线按年导出 Parquet(默认全部)
|
||||
```
|
||||
|
||||
说明:
|
||||
|
||||
Reference in New Issue
Block a user