- 移除 client.py 和 daily.py 中的调试日志 - 重构 rate_limiter 支持无限超时和更精确的令牌获取 - 变更 stock_basic 存储方案 HDF5 → CSV - 更新项目规则:强制使用 uv、禁止读取 config/ 目录 - 新增数据同步模块 sync.py 和测试 - .gitignore 添加 !data/ 允许跟踪数据文件
6.6 KiB
6.6 KiB
ProStock 代理指南
A股量化投资框架 - Python 项目,用于量化股票投资分析。
构建/检查/测试命令
⚠️ 重要:本项目强制使用 uv 作为 Python 包管理器和运行工具。禁止直接使用 python 或 pip 命令。
# 安装依赖(必须使用 uv)
uv pip install -e .
# 运行所有测试
uv run pytest
# 运行单个测试文件
uv run pytest tests/test_sync.py
# 运行单个测试类
uv run pytest tests/test_sync.py::TestDataSync
# 运行单个测试方法
uv run pytest tests/test_sync.py::TestDataSync::test_get_all_stock_codes_from_daily
# 使用详细输出运行
uv run pytest -v
# 运行覆盖率测试(如果安装了 pytest-cov)
uv run pytest --cov=src --cov-report=term-missing
禁止的命令 ❌
以下命令在本项目中严格禁止:
# 禁止直接使用 python
python -c "..." # 禁止!
python script.py # 禁止!
python -m pytest # 禁止!
python -m pip install # 禁止!
# 禁止直接使用 pip
pip install -e . # 禁止!
pip install package # 禁止!
pip list # 禁止!
正确的 uv 用法 ✅
# 运行 Python 代码
uv run python -c "..." # ✅ 正确
uv run python script.py # ✅ 正确
# 安装依赖
uv pip install -e . # ✅ 正确
uv pip install package # ✅ 正确
# 运行测试
uv run pytest # ✅ 正确
uv run pytest tests/test_sync.py # ✅ 正确
项目结构
ProStock/
├── src/ # 源代码
│ ├── data/ # 数据采集模块
│ │ ├── __init__.py
│ │ ├── client.py # Tushare API 客户端,带速率限制
│ │ ├── config.py # 配置(pydantic-settings)
│ │ ├── daily.py # 日线市场数据
│ │ ├── rate_limiter.py # 令牌桶速率限制器
│ │ ├── stock_basic.py # 股票基本信息
│ │ ├── storage.py # HDF5 存储管理器
│ │ └── sync.py # 数据同步
│ ├── config/ # 全局配置
│ │ ├── __init__.py
│ │ └── settings.py # 应用设置(pydantic-settings)
│ └── __init__.py
├── tests/ # 测试文件
│ ├── test_sync.py
│ └── test_daily.py
├── config/ # 配置文件
│ └── .env.local # 环境变量(不在 git 中)
├── data/ # 数据存储(HDF5 文件)
├── docs/ # 文档
├── pyproject.toml # 项目配置
└── README.md
代码风格指南
Python 版本
- 需要 Python 3.10+
- 使用现代 Python 特性(match/case、海象运算符、类型提示)
导入
# 标准库优先
import os
import time
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional, Dict, Callable
from concurrent.futures import ThreadPoolExecutor
import threading
# 第三方包
import pandas as pd
import numpy as np
from tqdm import tqdm
from pydantic_settings import BaseSettings
# 本地模块(使用来自 src 的绝对导入)
from src.data.client import TushareClient
from src.data.storage import Storage
from src.config.settings import get_settings
类型提示
- 始终使用类型提示 用于函数参数和返回值
- 对可空类型使用
Optional[X] - 当可用时使用现代联合语法
X | Y(Python 3.10+) - 从
typing导入类型:Optional、Dict、Callable等
def sync_single_stock(
self,
ts_code: str,
start_date: str,
end_date: str,
) -> pd.DataFrame:
...
文档字符串
- 使用 Google 风格文档字符串
- 包含 Args、Returns 部分
- 第一行保持简短摘要
def get_next_date(date_str: str) -> str:
"""获取给定日期之后的下一天。
Args:
date_str: YYYYMMDD 格式的日期
Returns:
YYYYMMDD 格式的下一天日期
"""
...
命名约定
- 变量、函数、方法使用
snake_case - 类使用
PascalCase - 常量使用
UPPER_CASE - 私有方法:
_leading_underscore - 受保护属性:
_single_underscore
错误处理
- 使用特定的异常,不要使用裸
except: - 使用上下文记录错误:
print(f"[ERROR] 上下文: {e}") - 对 API 调用使用指数退避重试逻辑
- 在关键错误时立即停止(设置停止标志)
try:
data = api.query(...)
except Exception as e:
print(f"[ERROR] 获取 {ts_code} 失败: {e}")
raise # 记录后重新抛出
配置
- 对所有配置使用 pydantic-settings
- 从
config/.env.local文件加载 - 环境变量自动转换:
tushare_token→TUSHARE_TOKEN - 对配置单例使用
@lru_cache()
数据存储
- 通过
pandas.HDFStore使用 HDF5 格式 进行持久化 - 存储在
data/目录中(通过DATA_PATH环境变量配置) - 对可追加数据集使用
format="table" - 追加时处理重复项:
drop_duplicates(subset=[...])
线程与并发
- 对 I/O 密集型任务(API 调用)使用
ThreadPoolExecutor - 实现停止标志以实现优雅关闭:
threading.Event() - 数据同步默认工作线程数:10
- 出错时始终使用
executor.shutdown(wait=False, cancel_futures=True)
日志记录
- 使用带前缀的 print 语句:
[模块名] 消息 - 错误格式:
[ERROR] 上下文: 异常 - 进度:循环中使用
tqdm
测试
- 使用 pytest 框架
- 模拟外部依赖(Tushare API)
- 使用
@pytest.fixture进行测试设置 - 在导入位置打补丁:
patch('src.data.sync.Storage') - 测试成功和错误两种情况
日期格式
- 使用
YYYYMMDD字符串格式表示日期 - 辅助函数:
get_today_date()、get_next_date() - 完全同步的默认开始日期:
20180101
依赖项
关键包:
pandas>=2.0.0- 数据处理numpy>=1.24.0- 数值计算tushare>=2.0.0- A股数据 APIpydantic>=2.0.0、pydantic-settings>=2.0.0- 配置tqdm>=4.65.0- 进度条pytest- 测试(开发)
环境变量
创建 config/.env.local:
TUSHARE_TOKEN=your_token_here
DATA_PATH=data
RATE_LIMIT=100
THREADS=10
常见任务
# 同步所有股票(增量)
uv run python -c "from src.data.sync import sync_all; sync_all()"
# 强制完全同步
uv run python -c "from src.data.sync import sync_all; sync_all(force_full=True)"
# 自定义线程数
uv run python -c "from src.data.sync import sync_all; sync_all(max_workers=20)"