57 lines
2.0 KiB
Python
57 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from _bootstrap import add_src_to_path
|
|
|
|
add_src_to_path()
|
|
|
|
from lhbfx.pipeline import import_range
|
|
|
|
|
|
def log_line(log_file: Path, message: str) -> None:
|
|
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
line = f"[{timestamp}] {message}"
|
|
print(line, flush=True)
|
|
log_file.parent.mkdir(parents=True, exist_ok=True)
|
|
with log_file.open("a", encoding="utf-8") as f:
|
|
f.write(line + "\n")
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="按工作日批量导入某年同花顺龙虎榜数据")
|
|
parser.add_argument("--year", type=int, default=2026, help="年份，默认 2026")
|
|
parser.add_argument("--start", default=None, help="起始日期 YYYY-MM-DD")
|
|
parser.add_argument("--end", default=None, help="结束日期 YYYY-MM-DD")
|
|
parser.add_argument("--limit", type=int, default=None, help="限制导入的交易日数量")
|
|
parser.add_argument("--log-file", default=None, help="日志文件路径")
|
|
args = parser.parse_args()
|
|
|
|
default_log = Path(__file__).resolve().parents[1] / "output" / "logs" / f"import_{args.year}.log"
|
|
log_file = Path(args.log_file) if args.log_file else default_log
|
|
|
|
log_line(
|
|
log_file,
|
|
f"开始导入 year={args.year}, start={args.start}, end={args.end}, limit={args.limit}",
|
|
)
|
|
result = import_range(year=args.year, start=args.start, end=args.end, limit=args.limit)
|
|
|
|
for item in result["success"]:
|
|
log_line(
|
|
log_file,
|
|
f"{item['trade_date']} -> 成功, 概览 {item['overview_count']} 条, 明细 {item['detail_count']} 条",
|
|
)
|
|
for item in result["failed"]:
|
|
log_line(log_file, f"{item['trade_date']} -> 失败, error={item['error']}")
|
|
|
|
log_line(
|
|
log_file,
|
|
f"导入完成 success={result['success_count']}, failed={result['failed_count']}, total={result['requested_days']}",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|