diff --git a/.gitignore b/.gitignore index 3996e81..71115b7 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,6 @@ +# 项目配置（含密码等敏感信息） +config.yaml + # ---> Python # Byte-compiled / optimized / DLL files __pycache__/ @@ -174,3 +177,4 @@ cython_debug/ # PyPI configuration file .pypirc +.idea/ diff --git a/.idea/.gitignore b/.idea/.gitignore new file mode 100644 index 0000000..ce1d3ea --- /dev/null +++ b/.idea/.gitignore @@ -0,0 +1,10 @@ +# 默认忽略的文件 +/shelf/ +/workspace.xml +# 基于编辑器的 HTTP 客户端请求 +/httpRequests/ +# 已忽略包含查询文件的默认文件夹 +/queries/ +# Datasource local storage ignored files +/dataSources/ +/dataSources.local.xml diff --git a/.idea/ashare-data.iml b/.idea/ashare-data.iml new file mode 100644 index 0000000..ed39dea --- /dev/null +++ b/.idea/ashare-data.iml @@ -0,0 +1,8 @@ + + + + + + + + \ No newline at end of file diff --git a/.idea/claudeCodeTabState.xml b/.idea/claudeCodeTabState.xml new file mode 100644 index 0000000..5ca02c5 --- /dev/null +++ b/.idea/claudeCodeTabState.xml @@ -0,0 +1,21 @@ + + + + + + \ No newline at end of file diff --git a/.idea/inspectionProfiles/profiles_settings.xml b/.idea/inspectionProfiles/profiles_settings.xml new file mode 100644 index 0000000..449e696 --- /dev/null +++ b/.idea/inspectionProfiles/profiles_settings.xml @@ -0,0 +1,6 @@ + + + + \ No newline at end of file diff --git a/.idea/modules.xml b/.idea/modules.xml new file mode 100644 index 0000000..e5a82cf --- /dev/null +++ b/.idea/modules.xml @@ -0,0 +1,8 @@ + + + + + + + + \ No newline at end of file diff --git a/README.md b/README.md index 33a33af..3c666bf 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,263 @@ # ashare-data -保存A股数据 \ No newline at end of file +A股数据抓取工具，使用 [AKShare](https://github.com/akfamily/akshare) 获取数据，保存到 MySQL 数据库。 + +## 功能概览 + +| 数据类型 | 说明 | 数据来源 | +|---------|------|---------| +| 股票列表 | A股全部股票代码和名称 | 东方财富 | +| 日线行情 | 开盘价、收盘价、最高价、最低价、成交量、成交额、涨跌幅等（前复权） | 东方财富 | +| 财务报表 | 利润表、资产负债表、现金流量表 | 新浪财经 | +| 资金流向 | 主力/超大/大/中/小单净流入及占比 | 东方财富 | +| 龙虎榜 | 上榜股票、买入卖出金额、上榜原因 | 东方财富 | +| 分红送转 | 每10股送转、派息、股息率等 | 巨潮资讯 | + +## 快速开始 + +### 1. 环境要求 + +- Python >= 3.10 +- MySQL >= 5.7（建议 8.0+） + +### 2. 安装依赖 + +```bash +python -m venv .venv +.venv\Scripts\activate # Windows +# source .venv/bin/activate # Linux/Mac + +pip install -r requirements.txt +``` + +### 3. 配置数据库 + +复制配置文件并修改 MySQL 连接信息： + +```bash +cp config.example.yaml config.yaml +``` + +编辑 `config.yaml`，填写你的 MySQL 连接信息： + +```yaml +mysql: + host: "localhost" + port: 3306 + user: "root" + password: "your_password" + database: "ashare" + charset: "utf8mb4" + +fetch: + delay: 0.5 # 请求间隔(秒)，防止被限流 + retry: 3 # 失败重试次数 +``` + +确保 MySQL 中已创建对应数据库： + +```sql +CREATE DATABASE ashare CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; +``` + +### 4. 运行 + +```bash +# 先抓取股票列表（其他模块依赖此数据） +python -m src.main --stock-list + +# 抓取最近30天的日线行情 +python -m src.main --daily + +# 指定日期范围抓取日线 +python -m src.main --daily --start-date 20260430 --end-date 20260508 + +# 抓取财务报表（全部股票） +python -m src.main --financial + +# 抓取单只股票的财务数据 +python -m src.main --financial --symbol 000001 + +# 抓取资金流向 +python -m src.main --money-flow + +# 抓取最近30天的龙虎榜 +python -m src.main --dragon-tiger + +# 指定日期范围抓取龙虎榜 +python -m src.main --dragon-tiger --start-date 20260423 --end-date 20250509 + +# 抓取分红送转 +python -m src.main --dividend + +# 全量抓取所有数据 +python -m src.main --all +``` + +### 5. 命令行参数说明 + +``` +usage: main.py [-h] [--stock-list] [--daily] [--financial] [--money-flow] + [--dragon-tiger] [--dividend] [--all] + [--start-date START_DATE] [--end-date END_DATE] + [--symbol SYMBOL] + +选项: + --stock-list 抓取A股股票列表 + --daily 抓取日线行情 + --financial 抓取财务报表（利润表、资产负债表、现金流量表） + --money-flow 抓取个股资金流向 + --dragon-tiger 抓取龙虎榜数据 + --dividend 抓取分红送转数据 + --all 全量抓取以上所有数据 + +日期过滤（仅对日线行情和龙虎榜生效）: + --start-date 开始日期，格式 YYYYMMDD，默认30天前 + --end-date 结束日期，格式 YYYYMMDD，默认今天 + +股票过滤（仅对财务报表和分红送转生效）: + --symbol 指定单只股票代码，如 000001，默认全部股票 +``` + +## 数据库表结构 + +### stock_info — 股票基本信息 + +| 字段 | 类型 | 说明 | +|------|------|------| +| code | VARCHAR(10) PK | 股票代码 | +| name | VARCHAR(50) | 股票名称 | + +### stock_daily — 日线行情（前复权） + +| 字段 | 类型 | 说明 | +|------|------|------| +| code | VARCHAR(10) | 股票代码 | +| date | DATE | 交易日期 | +| open | FLOAT | 开盘价 | +| close | FLOAT | 收盘价 | +| high | FLOAT | 最高价 | +| low | FLOAT | 最低价 | +| volume | FLOAT | 成交量 | +| turnover | FLOAT | 成交额 | +| amplitude | FLOAT | 振幅% | +| pct_change | FLOAT | 涨跌幅% | +| change | FLOAT | 涨跌额 | +| turnover_rate | FLOAT | 换手率% | + +联合主键：`(code, date)` + +### stock_financial_income — 利润表 + +| 字段 | 类型 | 说明 | +|------|------|------| +| code | VARCHAR(10) | 股票代码 | +| report_date | VARCHAR(20) | 报告期 | +| data | TEXT | JSON格式利润表数据 | + +联合主键：`(code, report_date)` + +### stock_financial_balance — 资产负债表 + +| 字段 | 类型 | 说明 | +|------|------|------| +| code | VARCHAR(10) | 股票代码 | +| report_date | VARCHAR(20) | 报告期 | +| data | TEXT | JSON格式资产负债表数据 | + +联合主键：`(code, report_date)` + +### stock_financial_cashflow — 现金流量表 + +| 字段 | 类型 | 说明 | +|------|------|------| +| code | VARCHAR(10) | 股票代码 | +| report_date | VARCHAR(20) | 报告期 | +| data | TEXT | JSON格式现金流量表数据 | + +联合主键：`(code, report_date)` + +### stock_money_flow — 个股资金流向 + +| 字段 | 类型 | 说明 | +|------|------|------| +| code | VARCHAR(10) | 股票代码 | +| date | DATE | 日期 | +| close | FLOAT | 收盘价 | +| pct_change | FLOAT | 涨跌幅% | +| main_net_inflow | FLOAT | 主力净流入-净额 | +| main_net_pct | FLOAT | 主力净流入-净占比 | +| huge_net_inflow | FLOAT | 超大盘净流入-净额 | +| huge_net_pct | FLOAT | 超大盘净流入-净占比 | +| big_net_inflow | FLOAT | 大盘净流入-净额 | +| big_net_pct | FLOAT | 大盘净流入-净占比 | +| mid_net_inflow | FLOAT | 中盘净流入-净额 | +| mid_net_pct | FLOAT | 中盘净流入-净占比 | +| small_net_inflow | FLOAT | 小盘净流入-净额 | +| small_net_pct | FLOAT | 小盘净流入-净占比 | + +联合主键：`(code, date)` + +### stock_dragon_tiger — 龙虎榜 + +| 字段 | 类型 | 说明 | +|------|------|------| +| code | VARCHAR(10) | 股票代码 | +| name | VARCHAR(50) | 股票名称 | +| date | DATE | 上榜日期 | +| close | FLOAT | 收盘价 | +| pct_change | FLOAT | 涨跌幅% | +| reason | VARCHAR(200) | 上榜原因 | +| buy_amount | FLOAT | 买入额 | +| sell_amount | FLOAT | 卖出额 | +| net_amount | FLOAT | 净额 | + +联合主键：`(code, date)` + +### stock_dividend — 分红送转 + +| 字段 | 类型 | 说明 | +|------|------|------| +| code | VARCHAR(10) | 股票代码 | +| name | VARCHAR(50) | 股票名称 | +| report_date | VARCHAR(20) | 报告期 | +| dividend_date | DATE | 除权除息日 | +| bonus_ratio | FLOAT | 每10股送转比例 | +| cash_div | FLOAT | 每10股派息 | +| convert_ratio | FLOAT | 每10股转增比例 | +| ex_right_date | DATE | 除权日 | +| dividend_yield | FLOAT | 股息率% | + +联合主键：`(code, report_date)` + +## 项目结构 + +``` +ashare-data/ +├── config.example.yaml # 配置文件模板 +├── config.yaml # 实际配置（含密码，已加入.gitignore） +├── pyproject.toml # 项目元数据 +├── requirements.txt # Python依赖 +├── src/ +│ ├── __init__.py +│ ├── config.py # 配置读取模块 +│ ├── db.py # 数据库模型与连接管理 +│ ├── main.py # 命令行入口 +│ └── fetchers/ +│ ├── __init__.py +│ ├── stock_list.py # 股票列表抓取 +│ ├── daily.py # 日线行情抓取 +│ ├── financial.py # 财务报表抓取 +│ ├── money_flow.py # 资金流向抓取 +│ ├── dragon_tiger.py # 龙虎榜抓取 +│ └── dividend.py # 分红送转抓取 +└── README.md +``` + +## 设计说明 + +- **去重写入**：所有表使用 `INSERT ON DUPLICATE KEY UPDATE`（upsert），重复执行不会产生重复数据 +- **自动重试**：网络请求失败自动重试（默认3次），单只股票失败不影响整体 +- **限速保护**：请求间自动延迟（默认0.5秒），防止被数据源限流 +- **增量更新**：日线行情和龙虎榜支持通过 `--start-date` / `--end-date` 指定日期范围 +- **懒加载导入**：各fetcher模块按需导入，未使用的模块不会加载 diff --git a/config.example.yaml b/config.example.yaml new file mode 100644 index 0000000..ff7c092 --- /dev/null +++ b/config.example.yaml @@ -0,0 +1,13 @@ +mysql: + host: "localhost" + port: 3306 + user: "root" + password: "your_password" + database: "ashare" + charset: "utf8mb4" + +fetch: + # 请求间隔(秒)，避免被限流 + delay: 0.5 + # 失败重试次数 + retry: 3 diff --git a/config.yaml b/config.yaml new file mode 100644 index 0000000..0b8c9c5 --- /dev/null +++ b/config.yaml @@ -0,0 +1,14 @@ +# MySQL 数据库连接配置 +mysql: + host: "db.freeicu.top" # 数据库地址 + port: 32000 # 端口号 + user: "root" # 用户名 + password: "ttx2011" # 密码 + database: "ashare" # 数据库名（需提前创建） + charset: "utf8mb4" # 字符集，支持中文 + +# 数据抓取配置 +fetch: + delay: 0.1 # 每次请求间隔（秒） + retry: 2 # 失败重试次数 + workers: 5 # 并发线程数；3个数据源(BaoStock+新浪+腾讯)，建议 5-8 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..ff20c9d --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,15 @@ +[project] +name = "ashare-data" +version = "0.1.0" +description = "A股数据抓取，保存到MySQL数据库" +requires-python = ">=3.10" +dependencies = [ + "akshare", + "pymysql", + "sqlalchemy>=2.0", + "pyyaml", + "pandas", +] + +[project.scripts] +ashare = "src.main:main" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..aa4aee9 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +akshare +pymysql +sqlalchemy>=2.0 +pyyaml +pandas diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..473a0f4 diff --git a/src/config.py b/src/config.py new file mode 100644 index 0000000..119cc5d --- /dev/null +++ b/src/config.py @@ -0,0 +1,47 @@ +"""全局配置加载模块，读取 config.yaml 提供 MySQL 连接和抓取参数""" + +import os +import yaml +from pathlib import Path + +_CONFIG = None + + +def load_config(config_path: str | None = None) -> dict: + global _CONFIG + if _CONFIG is not None: + return _CONFIG + + # 支持通过环境变量 ASHARE_CONFIG 指定配置文件路径 + if config_path is None: + config_path = os.environ.get("ASHARE_CONFIG", "config.yaml") + + path = Path(config_path) + if not path.exists(): + raise FileNotFoundError( + f"配置文件不存在: {path}\n" + f"请复制 config.example.yaml 为 config.yaml 并填写配置" + ) + + with open(path, "r", encoding="utf-8") as f: + _CONFIG = yaml.safe_load(f) + return _CONFIG + + +def get_mysql_url(config: dict | None = None) -> str: + """构建 SQLAlchemy 连接 URL，使用 pymysql 驱动""" + if config is None: + config = load_config() + m = config["mysql"] + return ( + f"mysql+pymysql://{m['user']}:{m['password']}" + f"@{m['host']}:{m['port']}/{m['database']}" + f"?charset={m.get('charset', 'utf8mb4')}" + ) + + +def get_fetch_config(config: dict | None = None) -> dict: + """返回抓取相关配置（delay/retry/workers），缺失时使用默认值""" + if config is None: + config = load_config() + return config.get("fetch", {"delay": 0.5, "retry": 3}) diff --git a/src/db.py b/src/db.py new file mode 100644 index 0000000..d178b3d --- /dev/null +++ b/src/db.py @@ -0,0 +1,271 @@ +"""数据库模型定义与连接管理 + +表结构概览： + - stock_info: 股票基本信息（含上市日期，用于跳过未上市股票） + - stock_daily: 日线行情（多源抓取：BaoStock/新浪/腾讯） + - stock_financial_income/balance/cashflow: 三大财务报表（JSON存储） + - stock_money_flow: 个股资金流向 + - stock_dragon_tiger: 龙虎榜 + - stock_dividend: 分红送转 + - stock_intraday: 1分钟分时行情 + - trading_day: 交易日历（用于判断数据完整性） + - stock_no_data: 无数据/停牌记录（避免重复抓取） +""" + +from sqlalchemy import ( + Column, String, Date, DateTime, Float, Integer, Text, + UniqueConstraint, Index, create_engine, MetaData, func, text, +) +from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker +from sqlalchemy.dialects.mysql import insert as mysql_insert + +from src.config import get_mysql_url + + +class Base(DeclarativeBase): + pass + + +# ── 股票基本信息 ── +class StockInfo(Base): + __tablename__ = "stock_info" + + code = Column(String(10), primary_key=True, comment="股票代码") + name = Column(String(50), comment="股票名称") + # 上市日期用于在抓取历史数据时跳过当时尚未上市的股票 + ipo_date = Column(Date, comment="上市日期") + + +# ── 日线行情 ── +class StockDaily(Base): + __tablename__ = "stock_daily" + __table_args__ = ( + UniqueConstraint("code", "date", name="uq_daily_code_date"), + Index("ix_daily_date", "date"), + ) + + id = Column(Integer, primary_key=True, autoincrement=True) + code = Column(String(10), nullable=False, comment="股票代码") + # 多个数据源（BaoStock/新浪/腾讯）写入同一张表，通过 upsert 去重 + date = Column(Date, nullable=False, comment="交易日期") + open = Column(Float, comment="开盘价") + close = Column(Float, comment="收盘价") + high = Column(Float, comment="最高价") + low = Column(Float, comment="最低价") + volume = Column(Float, comment="成交量") + turnover = Column(Float, comment="成交额") + amplitude = Column(Float, comment="振幅%") + pct_change = Column(Float, comment="涨跌幅%") + change = Column(Float, comment="涨跌额") + turnover_rate = Column(Float, comment="换手率%") + + +# ── 利润表 ── +class FinancialIncome(Base): + __tablename__ = "stock_financial_income" + __table_args__ = ( + UniqueConstraint("code", "report_date", name="uq_income_code_date"), + ) + + id = Column(Integer, primary_key=True, autoincrement=True) + code = Column(String(10), nullable=False, comment="股票代码") + report_date = Column(String(20), nullable=False, comment="报告期") + data = Column(Text, comment="JSON格式利润表数据") + + +# ── 资产负债表 ── +class FinancialBalance(Base): + __tablename__ = "stock_financial_balance" + __table_args__ = ( + UniqueConstraint("code", "report_date", name="uq_balance_code_date"), + ) + + id = Column(Integer, primary_key=True, autoincrement=True) + code = Column(String(10), nullable=False, comment="股票代码") + report_date = Column(String(20), nullable=False, comment="报告期") + data = Column(Text, comment="JSON格式资产负债表数据") + + +# ── 现金流量表 ── +class FinancialCashflow(Base): + __tablename__ = "stock_financial_cashflow" + __table_args__ = ( + UniqueConstraint("code", "report_date", name="uq_cashflow_code_date"), + ) + + id = Column(Integer, primary_key=True, autoincrement=True) + code = Column(String(10), nullable=False, comment="股票代码") + report_date = Column(String(20), nullable=False, comment="报告期") + data = Column(Text, comment="JSON格式现金流量表数据") + + +# ── 资金流向 ── +class StockMoneyFlow(Base): + __tablename__ = "stock_money_flow" + __table_args__ = ( + UniqueConstraint("code", "date", name="uq_moneyflow_code_date"), + Index("ix_moneyflow_date", "date"), + ) + + id = Column(Integer, primary_key=True, autoincrement=True) + code = Column(String(10), nullable=False, comment="股票代码") + date = Column(Date, nullable=False, comment="日期") + close = Column(Float, comment="收盘价") + pct_change = Column(Float, comment="涨跌幅%") + main_net_inflow = Column(Float, comment="主力净流入-净额") + main_net_pct = Column(Float, comment="主力净流入-净占比") + huge_net_inflow = Column(Float, comment="超大盘净流入-净额") + huge_net_pct = Column(Float, comment="超大盘净流入-净占比") + big_net_inflow = Column(Float, comment="大盘净流入-净额") + big_net_pct = Column(Float, comment="大盘净流入-净占比") + mid_net_inflow = Column(Float, comment="中盘净流入-净额") + mid_net_pct = Column(Float, comment="中盘净流入-净占比") + small_net_inflow = Column(Float, comment="小盘净流入-净额") + small_net_pct = Column(Float, comment="小盘净流入-净占比") + + +# ── 龙虎榜 ── +class StockDragonTiger(Base): + __tablename__ = "stock_dragon_tiger" + __table_args__ = ( + UniqueConstraint("code", "date", name="uq_lhb_code_date"), + Index("ix_lhb_date", "date"), + ) + + id = Column(Integer, primary_key=True, autoincrement=True) + code = Column(String(10), nullable=False, comment="股票代码") + name = Column(String(50), comment="股票名称") + date = Column(Date, nullable=False, comment="上榜日期") + close = Column(Float, comment="收盘价") + pct_change = Column(Float, comment="涨跌幅%") + reason = Column(String(200), comment="上榜原因") + buy_amount = Column(Float, comment="买入额") + sell_amount = Column(Float, comment="卖出额") + net_amount = Column(Float, comment="净额") + + +# ── 分红送转 ── +class StockDividend(Base): + __tablename__ = "stock_dividend" + __table_args__ = ( + UniqueConstraint("code", "report_date", name="uq_dividend_code_date"), + ) + + id = Column(Integer, primary_key=True, autoincrement=True) + code = Column(String(10), nullable=False, comment="股票代码") + name = Column(String(50), comment="股票名称") + report_date = Column(String(20), nullable=False, comment="报告期") + dividend_date = Column(Date, comment="除权除息日") + bonus_ratio = Column(Float, comment="每10股送转比例") + cash_div = Column(Float, comment="每10股派息") + convert_ratio = Column(Float, comment="每10股转增比例") + ex_right_date = Column(Date, comment="除权日") + dividend_yield = Column(Float, comment="股息率%") + + +# ── 交易日历 ── +class TradingDay(Base): + __tablename__ = "trading_day" + __table_args__ = ( + UniqueConstraint("date", name="uq_trading_day_date"), + ) + + id = Column(Integer, primary_key=True, autoincrement=True) + date = Column(Date, nullable=False, comment="交易日期") + + +# ── 无数据/停牌记录（按天粒度） ── +class StockNoData(Base): + __tablename__ = "stock_no_data" + __table_args__ = ( + UniqueConstraint("code", "date", name="uq_nodata_code_date"), + ) + + id = Column(Integer, primary_key=True, autoincrement=True) + code = Column(String(10), nullable=False, comment="股票代码") + date = Column(Date, nullable=False, comment="停牌/无数据日期") + created_at = Column(DateTime, server_default=func.now(), comment="记录时间") + + +# ── 分时行情（1分钟线） ── +class StockIntraday(Base): + __tablename__ = "stock_intraday" + __table_args__ = ( + UniqueConstraint("code", "datetime", name="uq_intraday_code_dt"), + Index("ix_intraday_date", "datetime"), + ) + + id = Column(Integer, primary_key=True, autoincrement=True) + code = Column(String(10), nullable=False, comment="股票代码") + datetime = Column(DateTime, nullable=False, comment="时间") + open = Column(Float, comment="开盘价") + high = Column(Float, comment="最高价") + low = Column(Float, comment="最低价") + close = Column(Float, comment="收盘价") + volume = Column(Float, comment="成交量") + amount = Column(Float, comment="成交额") + + +# ── 数据库连接管理 ── +_engine = None +_SessionFactory = None + + +def get_engine(): + global _engine + if _engine is None: + _engine = create_engine(get_mysql_url(), pool_size=5, pool_recycle=3600) + return _engine + + +def get_session() -> Session: + global _SessionFactory + if _SessionFactory is None: + _SessionFactory = sessionmaker(bind=get_engine()) + return _SessionFactory() + + +def init_db(): + engine = get_engine() + # 自动迁移：旧版 stock_no_data 使用 date_range 列，新版改为 date 列 + # 检测到旧表结构时先删除，由 create_all 重建 + with engine.connect() as conn: + result = conn.execute(text("SHOW COLUMNS FROM stock_no_data LIKE 'date_range'")) + if result.fetchone(): + conn.execute(text("DROP TABLE stock_no_data")) + conn.commit() + print(" stock_no_data 表结构已升级（date_range → date）") + Base.metadata.create_all(engine) + print("数据库表初始化完成") + + +def batch_upsert(model_cls: type[Base], rows: list[dict], index_columns: list[str]): + """MySQL批量upsert：INSERT ON DUPLICATE KEY UPDATE + + index_columns: 用于判断重复的唯一键列名（如 ["code", "date"]）， + 这些列在冲突时不更新，其余列使用新值覆盖。 + """ + if not rows: + return + session = get_session() + try: + stmt = mysql_insert(model_cls).values(rows) + # 只更新输入数据中实际包含的列，排除唯一键列、自增主键和 server_default 列 + input_keys = set(rows[0].keys()) + update_dict = { + col.name: stmt.inserted[col.name] + for col in model_cls.__table__.columns + if col.name in input_keys + and col.name not in index_columns + and not col.primary_key + and not col.server_default + } + if update_dict: + stmt = stmt.on_duplicate_key_update(**update_dict) + session.execute(stmt) + session.commit() + except Exception as e: + session.rollback() + raise e + finally: + session.close() diff --git a/src/fetchers/__init__.py b/src/fetchers/__init__.py new file mode 100644 index 0000000..473a0f4 diff --git a/src/fetchers/daily.py b/src/fetchers/daily.py new file mode 100644 index 0000000..b845b34 --- /dev/null +++ b/src/fetchers/daily.py @@ -0,0 +1,398 @@ +"""日线行情抓取模块 — 核心模块，采用三数据源轮询 + 自动降级架构 + +数据源优先级（按速度排序）： + 1. BaoStock: 速度极快（~0.04s/只），但不支持北交所（920xxx），线程不安全需加锁 + 2. 新浪: 支持全部交易所（含北交所 bj 前缀），返回 JSONP 需解析 + 3. 腾讯: 不支持北交所，返回标准 JSON + +跳过策略（停牌天按天记录）： + - 数据完整 = 行情记录数 + 已标记停牌天数 >= 交易日总数 + - 未上市股票（ipo_date > 查询结束日期） + - 抓取成功后自动识别缺失交易日并标记为停牌 +""" + +import json +import re +import time +from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import datetime, timedelta +import requests +from src.config import get_fetch_config +from src.db import StockInfo, StockDaily, StockNoData, batch_upsert, get_session +from sqlalchemy import select, func + + +# 新浪日线接口：返回 JSONP 格式，需正则提取 JSON 数组 +_SINA_URL = "https://quotes.sina.cn/cn/api/jsonp_v2.php/var=/CN_MarketDataService.getKLineData" +# 腾讯日线接口：返回标准 JSON，支持前复权 +_TENCENT_URL = "https://web.ifzq.gtimg.cn/appstock/app/fqkline/get" + +_HEADERS = { + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", + "Referer": "https://finance.sina.com.cn", +} + +# 复用 HTTP Session 以利用连接池和 keep-alive，减少 TCP 握手开销 +_sina_session = requests.Session() +_sina_session.headers.update(_HEADERS) +_tencent_session = requests.Session() +_tencent_session.headers.update(_HEADERS) + +# BaoStock 全局连接：bs.login() 只需调用一次，但 query_history_k_data_plus 非线程安全 +import baostock as bs +import threading +_bs_lock = threading.Lock() +_bs_logged_in = False + + +def _bs_ensure_login(): + global _bs_logged_in + with _bs_lock: + if not _bs_logged_in: + bs.login() + _bs_logged_in = True + + +def _get_stock_codes() -> list[str]: + session = get_session() + try: + result = session.execute(select(StockInfo.code)) + return [row[0] for row in result] + finally: + session.close() + + +def _get_not_listed(end_date: str) -> set[str]: + """查询在end_date之后上市的股票（未上市，需跳过） + 例如抓取 20260501~20260508 的数据时，5月10日上市的股票应被跳过 + """ + session = get_session() + try: + result = session.execute( + select(StockInfo.code).where(StockInfo.ipo_date > end_date) + ) + return {row[0] for row in result} + finally: + session.close() + + +def _get_complete_codes(start_date: str, end_date: str, trading_days: list[str]) -> set[str]: + """数据完整的判断：行情记录数 + 已标记停牌天数 >= 交易日总数 + + 某只股票在范围内停牌了2天、有行情8天、共10个交易日 → 8+2=10, 视为完整。 + 这样停牌股票不会被反复抓取，同时部分停牌也能正确处理。 + """ + if not trading_days: + return set() + td_count = len(trading_days) + session = get_session() + try: + # 行情记录数 + rec_rows = session.execute( + select(StockDaily.code, func.count(StockDaily.id)) + .where(StockDaily.date >= start_date) + .where(StockDaily.date <= end_date) + .group_by(StockDaily.code) + ) + rec_counts = {row[0]: row[1] for row in rec_rows} + + # 停牌天数 + susp_rows = session.execute( + select(StockNoData.code, func.count(StockNoData.id)) + .where(StockNoData.date >= start_date) + .where(StockNoData.date <= end_date) + .group_by(StockNoData.code) + ) + susp_counts = {row[0]: row[1] for row in susp_rows} + + complete = set() + for code in set(rec_counts) | set(susp_counts): + if rec_counts.get(code, 0) + susp_counts.get(code, 0) >= td_count: + complete.add(code) + return complete + finally: + session.close() + + +def _record_nodata_days(code: str, days: list[str]): + """记录个股的停牌/无数据日期（按天粒度） + + 抓取成功后，对比实际交易日列表，缺失的日期即为停牌天。 + 全部源无数据时，所有交易日都标记为停牌。 + """ + if not days: + return + rows = [{"code": code, "date": d} for d in days] + batch_upsert(StockNoData, rows, ["code", "date"]) + + +def _code_to_prefix(code: str) -> str: + """转为新浪/腾讯接口的代码前缀格式（如 sh600000、sz000001）""" + # 北交所920开头需用 bj 前缀（新浪特有），否则返回数据中缺少日期字段 + if code.startswith("920"): + return f"bj{code}" + if code.startswith(("6", "9")): + return f"sh{code}" + return f"sz{code}" + + +def _code_to_baostock(code: str) -> str: + """BaoStock格式: sh.600000 / sz.000001""" + if code.startswith(("6", "9")): + return f"sh.{code}" + return f"sz.{code}" + + +def _clean(val): + """将空字符串、无效值转为 None + + BaoStock 对无数据的字段返回空字符串 ''，MySQL FLOAT 列不接受空字符串， + 不做转换会触发 DataError。此函数统一处理所有数据源的空值情况。 + """ + if val is None: + return None + if isinstance(val, str) and val.strip() == "": + return None + return val + + +def _fetch_sina(code: str, start_date: str, end_date: str, datalen: int) -> list[dict] | None: + """新浪数据源：支持全部交易所（含北交所 bj 前缀） + + 接口返回 JSONP 格式 `var=(...)`, 需正则提取 JSON 数组。 + datalen 参数控制返回的K线条数，新浪不支持精确日期范围过滤， + 所以拿到数据后再在客户端按日期范围筛选。 + """ + symbol = _code_to_prefix(code) + try: + r = _sina_session.get( + _SINA_URL, + params={"symbol": symbol, "scale": "240", "ma": "no", "datalen": str(datalen)}, + timeout=10, + ) + m = re.search(r"\((\[.*\])\)", r.text, re.DOTALL) + if not m: + return None + data = json.loads(m.group(1)) + if not data: + return None + # 新浪接口返回的日期格式为 "2026-05-08"，需转为统一格式做范围比较 + sd = f"{start_date[:4]}-{start_date[4:6]}-{start_date[6:8]}" + ed = f"{end_date[:4]}-{end_date[4:6]}-{end_date[6:8]}" + rows = [] + for item in data: + day = item.get("day", "") + if sd <= day <= ed: + rows.append({ + "code": code, "date": day, + "open": _clean(item.get("open")), "high": _clean(item.get("high")), + "low": _clean(item.get("low")), "close": _clean(item.get("close")), + "volume": _clean(item.get("volume")), "turnover": None, + "amplitude": None, "pct_change": None, "change": None, "turnover_rate": None, + }) + return rows if rows else None + except Exception: + return None + + +def _fetch_tencent(code: str, start_date: str, end_date: str) -> list[dict] | None: + """腾讯数据源：不支持北交所（920xxx），返回标准 JSON + + 接口支持精确日期范围查询和前复权（qfq），数据路径为 data.{symbol}.qfqday + """ + # 腾讯接口无北交所数据，直接跳过避免无效请求 + if code.startswith("920"): + return None + symbol = _code_to_prefix(code) + sd = f"{start_date[:4]}-{start_date[4:6]}-{start_date[6:8]}" + ed = f"{end_date[:4]}-{end_date[4:6]}-{end_date[6:8]}" + try: + r = _tencent_session.get( + _TENCENT_URL, + params={"param": f"{symbol},day,{sd},{ed},300,qfq"}, + timeout=10, + ) + d = r.json() + klines = d.get("data", {}).get(symbol, {}).get("qfqday") + if not klines: + return None + rows = [] + for k in klines: + rows.append({ + "code": code, "date": k[0], + "open": _clean(k[1]), "close": _clean(k[2]), "high": _clean(k[3]), "low": _clean(k[4]), + "volume": _clean(k[5]), "turnover": None, + "amplitude": None, "pct_change": None, "change": None, "turnover_rate": None, + }) + return rows if rows else None + except Exception: + return None + + +def _fetch_baostock(code: str, start_date: str, end_date: str) -> list[dict] | None: + """BaoStock 数据源：速度极快（~0.04s/只），但不支持北交所 + + BaoStock 的 query_history_k_data_plus 不是线程安全的， + 必须在 _bs_lock 保护下串行调用，否则会出现数据错乱。 + """ + if code.startswith("920"): + return None + symbol = _code_to_baostock(code) + # BaoStock日期格式: 2026-05-01 + sd = f"{start_date[:4]}-{start_date[4:6]}-{start_date[6:8]}" + ed = f"{end_date[:4]}-{end_date[4:6]}-{end_date[6:8]}" + try: + with _bs_lock: + # adjustflag="2" 表示前复权，amount 为成交额 + rs = bs.query_history_k_data_plus( + symbol, "date,open,high,low,close,volume,amount", + start_date=sd, end_date=ed, frequency="d", adjustflag="2", + ) + rows = [] + while (rs.error_code == "0") and rs.next(): + r = rs.get_row_data() + rows.append({ + "code": code, "date": r[0], + "open": _clean(r[1]), "high": _clean(r[2]), "low": _clean(r[3]), "close": _clean(r[4]), + "volume": _clean(r[5]), "turnover": _clean(r[6]), + "amplitude": None, "pct_change": None, "change": None, "turnover_rate": None, + }) + return rows if rows else None + except Exception: + return None + + +# 数据源列表，按速度排序，轮询分配以分散请求压力 +_SOURCES = ["baostock", "sina", "tencent"] + +_FETCH_FNS = { + "baostock": lambda code, sd, ed, dl: _fetch_baostock(code, sd, ed), + "sina": _fetch_sina, + "tencent": lambda code, sd, ed, dl: _fetch_tencent(code, sd, ed), +} + + +def _fetch_one(code: str, start_date: str, end_date: str, source: str, datalen: int) -> dict: + """单只股票抓取：主源失败时自动按顺序尝试其他数据源（降级策略） + + 例如主源为 baostock，失败后会依次尝试 sina、tencent， + 全部失败则标记为 fail 或 no_data。 + """ + ordered = [source] + [s for s in _SOURCES if s != source] + tried = [] + for src in ordered: + fn = _FETCH_FNS[src] + rows = fn(code, start_date, end_date, datalen) + tried.append(src) + if rows is not None: + label = src if src == source else f"{src}(fallback)" + return {"code": code, "status": "ok", "rows": rows, "source": label} + + return {"code": code, "status": "fail", "source": "→".join(tried), + "no_data": all(s in tried for s in _SOURCES)} + + +def fetch_daily(start_date: str | None = None, end_date: str | None = None): + cfg = get_fetch_config() + delay = cfg.get("delay", 0.1) + workers = cfg.get("workers", 5) + + codes = _get_stock_codes() + if not codes: + print(" 无股票列表，请先运行 --stock-list", flush=True) + return + + if end_date is None: + end_date = datetime.now().strftime("%Y%m%d") + if start_date is None: + start_date = (datetime.now() - timedelta(days=30)).strftime("%Y%m%d") + + start_dt = datetime.strptime(start_date, "%Y%m%d") + end_dt = datetime.strptime(end_date, "%Y%m%d") + # 多取10天以防节假日偏移导致数据不足 + datalen = (end_dt - start_dt).days + 10 + + # BaoStock 需要先 login() 才能查询，全局只需一次 + _bs_ensure_login() + + # 获取交易日历 + from src.fetchers.trading_day import get_trading_days + trading_days = get_trading_days(start_date, end_date) + td_count = len(trading_days) + print(f" 交易日历: {start_date} ~ {end_date} 共 {td_count} 个交易日", flush=True) + + # 跳过过滤：数据完整（含已标记停牌天数）+ 未上市 + complete = _get_complete_codes(start_date, end_date, trading_days) + not_listed = _get_not_listed(end_date) + skip_set = complete | not_listed + if complete: + print(f" {len(complete)} 只股票数据已完整（含停牌天），跳过...", flush=True) + if not_listed: + print(f" {len(not_listed)} 只股票未上市，跳过...", flush=True) + codes = [c for c in codes if c not in skip_set] + + # 交易日集合，用于抓取后比对缺失日期 + trading_days_set = set(trading_days) + + total = len(codes) + if total == 0: + print(" 所有股票数据已完整，无需抓取", flush=True) + return + + success = 0 + fail = 0 + nodata_count = 0 + skipped = len(skip_set) + t_start = time.time() + + # 轮询分配数据源：将股票均匀分配到3个源，分散请求压力 + sources = [_SOURCES[i % 3] for i in range(total)] + + print(f"正在抓取日线行情 {start_date} ~ {end_date}，需抓取 {total} 只（跳过 {skipped} 只），" + f"{workers} 线程 × 3 源(BaoStock+新浪+腾讯)...", flush=True) + + with ThreadPoolExecutor(max_workers=workers) as pool: + futures = {} + for idx, code in enumerate(codes): + f = pool.submit(_fetch_one, code, start_date, end_date, sources[idx], datalen) + futures[f] = code + + done_count = 0 + for f in as_completed(futures): + done_count += 1 + result = f.result() + code = result["code"] + status = result["status"] + + if status == "ok": + try: + batch_upsert(StockDaily, result["rows"], ["code", "date"]) + success += 1 + # 比对返回日期与交易日，缺失的标记为停牌天 + returned_dates = {row["date"] for row in result["rows"]} + missing_days = [d for d in trading_days if d not in returned_dates] + if missing_days: + _record_nodata_days(code, missing_days) + except Exception as e: + print(f" {code} 写入失败: {e}", flush=True) + fail += 1 + else: + if result.get("no_data"): + # 三个数据源全部返回空数据，所有交易日标记为停牌 + _record_nodata_days(code, trading_days) + nodata_count += 1 + else: + fail += 1 + + total_elapsed = time.time() - t_start + avg = total_elapsed / done_count + eta = avg * (total - done_count) + + src = result.get("source", "") + print(f" [{done_count}/{total}] {code} [{src}] 成功:{success} 失败:{fail} " + f"已用时:{total_elapsed:.0f}s 预计剩余:{eta:.0f}s", flush=True) + + time.sleep(delay) + + total_time = time.time() - t_start + print(f" 日线行情抓取完成，成功:{success} 失败:{fail} 停牌:{nodata_count} 总耗时:{total_time:.1f}s", flush=True) diff --git a/src/fetchers/dividend.py b/src/fetchers/dividend.py new file mode 100644 index 0000000..7b663d3 --- /dev/null +++ b/src/fetchers/dividend.py @@ -0,0 +1,120 @@ +"""分红送转数据抓取模块 — 使用巨潮信息网（CNInfo）数据源 + +通过 AKShare 的 stock_dividend_cninfo 接口按股票代码逐个查询， +返回该股票历史所有分红记录（含送股、转增、派息等）。 +""" + +import time +from datetime import datetime +import akshare as ak +from src.config import get_fetch_config +from src.db import StockInfo, StockDividend, batch_upsert, get_session +from sqlalchemy import select + + +def _get_stock_codes() -> list[str]: + session = get_session() + try: + result = session.execute(select(StockInfo.code)) + return [row[0] for row in result] + finally: + session.close() + + +def fetch_dividend(symbol: str | None = None): + """抓取分红送转数据""" + cfg = get_fetch_config() + delay = cfg.get("delay", 1.0) + retry = cfg.get("retry", 5) + + if symbol: + codes = [symbol] + else: + codes = _get_stock_codes() + + if not codes: + print(" 无股票列表，请先运行 --stock-list", flush=True) + return + + total = len(codes) + success = 0 + fail = 0 + consecutive_fail = 0 + + print(f"正在抓取分红送转数据，共 {total} 只股票...", flush=True) + + for i, code in enumerate(codes): + df = None + for attempt in range(retry): + try: + df = ak.stock_dividend_cninfo(symbol=code) + consecutive_fail = 0 + break + except Exception as e: + wait = 3 * (attempt + 1) + if attempt < retry - 1: + print(f" [{i+1}/{total}] {code} 第{attempt+1}次重试，等待{wait}秒...", flush=True) + time.sleep(wait) + else: + print(f" [{i+1}/{total}] {code} 抓取失败: {e}", flush=True) + fail += 1 + consecutive_fail += 1 + + if consecutive_fail >= 5: + print(f" 连续{consecutive_fail}只失败，暂停60秒...", flush=True) + time.sleep(60) + consecutive_fail = 0 + + if df is None or df.empty: + time.sleep(delay) + continue + + rows = [] + for _, row in df.iterrows(): + rows.append({ + "code": code, + "name": str(row.get("名称", row.get("name", ""))), + "report_date": str(row.get("报告期", row.get("report_date", ""))), + "dividend_date": _safe_date(row, ["分红年度", "除权除息日", "dividend_date"]), + "bonus_ratio": _safe_float(row, ["送转比例", "每10股送转", "bonus_ratio"]), + "cash_div": _safe_float(row, ["每10股派息", "现金分红", "cash_div"]), + "convert_ratio": _safe_float(row, ["转增比例", "每10股转增", "convert_ratio"]), + "ex_right_date": _safe_date(row, ["除权日", "除权除息日", "ex_right_date"]), + "dividend_yield": _safe_float(row, ["股息率", "dividend_yield"]), + }) + + try: + batch_upsert(StockDividend, rows, ["code", "report_date"]) + success += 1 + except Exception as e: + print(f" [{i+1}/{total}] {code} 写入失败: {e}", flush=True) + fail += 1 + + if (i + 1) % 50 == 0: + print(f" [{i+1}/{total}] 进度... 成功:{success} 失败:{fail}", flush=True) + else: + print(f" [{i+1}/{total}] {code} OK", flush=True) + + time.sleep(delay) + + print(f" 分红送转抓取完成，成功:{success} 失败:{fail}", flush=True) + + +def _safe_float(row, keys: list[str]): + for key in keys: + val = row.get(key) + if val is not None: + try: + return float(val) + except (ValueError, TypeError): + return None + return None + + +def _safe_date(row, keys: list[str]): + """从 DataFrame 行中按多个候选列名提取日期值""" + for key in keys: + val = row.get(key) + if val is not None and str(val).strip(): + return str(val).strip() + return None diff --git a/src/fetchers/dragon_tiger.py b/src/fetchers/dragon_tiger.py new file mode 100644 index 0000000..cc353c9 --- /dev/null +++ b/src/fetchers/dragon_tiger.py @@ -0,0 +1,69 @@ +"""龙虎榜数据抓取模块 — 使用东方财富数据源 + +龙虎榜按日期范围查询，返回该期间内所有上榜股票的买卖详情。 +数据量相对较小（每天几十到上百条），通常一次请求即可获取全部数据。 +""" + +import time +from datetime import datetime, timedelta +import akshare as ak +from src.config import get_fetch_config +from src.db import StockDragonTiger, batch_upsert + + +def fetch_dragon_tiger(start_date: str | None = None, end_date: str | None = None): + """抓取龙虎榜数据 + start_date/end_date: YYYYMMDD格式 + """ + cfg = get_fetch_config() + retry = cfg.get("retry", 3) + + if end_date is None: + end_date = datetime.now().strftime("%Y%m%d") + if start_date is None: + start_date = (datetime.now() - timedelta(days=30)).strftime("%Y%m%d") + + print(f"正在抓取龙虎榜数据 {start_date} ~ {end_date}...", flush=True) + + for attempt in range(retry): + try: + df = ak.stock_lhb_detail_em(start_date=start_date, end_date=end_date) + break + except Exception as e: + if attempt < retry - 1: + print(f" 抓取失败({attempt + 1}/{retry}): {e}，等待重试...", flush=True) + time.sleep(2) + else: + raise + + if df is None or df.empty: + print(" 无龙虎榜数据", flush=True) + return + + rows = [] + for _, row in df.iterrows(): + rows.append({ + "code": str(row.get("代码", row.get("code", ""))), + "name": row.get("名称", row.get("name", "")), + "date": row.get("上榜日", row.get("date", "")), + "close": _safe_float(row, ["收盘价", "close"]), + "pct_change": _safe_float(row, ["涨跌幅", "pct_change"]), + "reason": str(row.get("上榜原因", row.get("reason", ""))), + "buy_amount": _safe_float(row, ["买入额", "buy_amount"]), + "sell_amount": _safe_float(row, ["卖出额", "sell_amount"]), + "net_amount": _safe_float(row, ["净额", "net_amount"]), + }) + + batch_upsert(StockDragonTiger, rows, ["code", "date"]) + print(f" 龙虎榜抓取完成，共 {len(rows)} 条记录", flush=True) + + +def _safe_float(row, keys: list[str]): + for key in keys: + val = row.get(key) + if val is not None: + try: + return float(val) + except (ValueError, TypeError): + return None + return None diff --git a/src/fetchers/financial.py b/src/fetchers/financial.py new file mode 100644 index 0000000..3875cdf --- /dev/null +++ b/src/fetchers/financial.py @@ -0,0 +1,137 @@ +"""财务报表抓取模块 — 利润表、资产负债表、现金流量表 + +使用 AKShare 的新浪财务数据接口，每个接口返回一个报告期对应的所有字段。 +由于不同股票的字段名可能变化，采用 JSON 格式存储完整数据而非逐字段建列。 + +防限流策略： + - 指数退避重试（3s、6s、9s...） + - 连续5只股票失败时暂停60秒（可能触发了限流） +""" + +import json +import time +from datetime import datetime +import akshare as ak +from src.config import get_fetch_config +from src.db import StockInfo, FinancialIncome, FinancialBalance, FinancialCashflow, batch_upsert, get_session +from sqlalchemy import select + + +def _get_stock_codes() -> list[str]: + session = get_session() + try: + result = session.execute(select(StockInfo.code)) + return [row[0] for row in result] + finally: + session.close() + + +def fetch_financial(symbol: str | None = None): + """抓取财务报表数据（利润表、资产负债表、现金流量表）""" + cfg = get_fetch_config() + delay = cfg.get("delay", 1.0) + retry = cfg.get("retry", 5) + + if symbol: + codes = [symbol] + else: + codes = _get_stock_codes() + + if not codes: + print(" 无股票列表，请先运行 --stock-list", flush=True) + return + + total = len(codes) + success = 0 + fail = 0 + consecutive_fail = 0 + + print(f"正在抓取财务报表，共 {total} 只股票...", flush=True) + + for i, code in enumerate(codes): + income_df = None + balance_df = None + cashflow_df = None + for attempt in range(retry): + try: + income_df = ak.stock_financial_report_sina(stock=code, symbol="利润表") + balance_df = ak.stock_financial_report_sina(stock=code, symbol="资产负债表") + cashflow_df = ak.stock_financial_report_sina(stock=code, symbol="现金流量表") + consecutive_fail = 0 + break + except Exception as e: + wait = 3 * (attempt + 1) + if attempt < retry - 1: + print(f" [{i+1}/{total}] {code} 第{attempt+1}次重试，等待{wait}秒...", flush=True) + time.sleep(wait) + else: + print(f" [{i+1}/{total}] {code} 抓取失败: {e}", flush=True) + fail += 1 + consecutive_fail += 1 + + # 连续多次失败说明可能被限流，暂停一段时间让限制重置 + if consecutive_fail >= 5: + print(f" 连续{consecutive_fail}只失败，暂停60秒...", flush=True) + time.sleep(60) + consecutive_fail = 0 + + try: + if income_df is not None and not income_df.empty: + income_rows = _parse_financial_df(code, income_df) + batch_upsert(FinancialIncome, income_rows, ["code", "report_date"]) + + if balance_df is not None and not balance_df.empty: + balance_rows = _parse_financial_df(code, balance_df) + batch_upsert(FinancialBalance, balance_rows, ["code", "report_date"]) + + if cashflow_df is not None and not cashflow_df.empty: + cashflow_rows = _parse_financial_df(code, cashflow_df) + batch_upsert(FinancialCashflow, cashflow_rows, ["code", "report_date"]) + + success += 1 + except Exception as e: + print(f" [{i+1}/{total}] {code} 写入失败: {e}", flush=True) + fail += 1 + + if (i + 1) % 50 == 0: + print(f" [{i+1}/{total}] 进度... 成功:{success} 失败:{fail}", flush=True) + else: + print(f" [{i+1}/{total}] {code} OK", flush=True) + + time.sleep(delay) + + print(f" 财务报表抓取完成，成功:{success} 失败:{fail}", flush=True) + + +def _parse_financial_df(code: str, df) -> list[dict]: + """将财务报表 DataFrame 转为数据库行，所有字段以 JSON 格式存储 + + 因为不同报表的字段名和数量差异很大，且可能随时间变化， + 不适合逐字段建列。JSON 存储保留了原始数据的完整性。 + """ + rows = [] + report_date_col = None + for col in df.columns: + if "报告期" in col or "报告日" in col or "date" in col.lower(): + report_date_col = col + break + + if report_date_col is None and len(df.columns) > 0: + report_date_col = df.columns[0] + + for _, row in df.iterrows(): + report_date = str(row[report_date_col]) + data_dict = {col: _safe_val(row[col]) for col in df.columns if col != report_date_col} + rows.append({ + "code": code, + "report_date": report_date, + "data": json.dumps(data_dict, ensure_ascii=False), + }) + return rows + + +def _safe_val(val): + """安全转换值，处理 NaN（float NaN 不等于自身的特性）""" + if val is None or (isinstance(val, float) and val != val): + return None + return val diff --git a/src/fetchers/intraday.py b/src/fetchers/intraday.py new file mode 100644 index 0000000..ec410fd --- /dev/null +++ b/src/fetchers/intraday.py @@ -0,0 +1,137 @@ +"""分时行情抓取模块（1分钟线）— 使用新浪数据源 + +注意：分时数据量非常大（每只股票每天约240条分钟记录）， +目前暂未启用此模块的抓取任务。 +""" + +import time +from datetime import datetime +import akshare as ak +from src.config import get_fetch_config +from src.db import StockInfo, StockIntraday, batch_upsert, get_session +from sqlalchemy import select, func + + +def _get_stock_codes() -> list[str]: + session = get_session() + try: + result = session.execute(select(StockInfo.code)) + return [row[0] for row in result] + finally: + session.close() + + +def _get_existing_codes() -> set[str]: + """查询已有分时数据的股票代码（最近5个交易日有数据的视为已有） + + 分时数据时效性强，只保留近期数据即可，避免重复抓取。 + """ + session = get_session() + try: + result = session.execute( + select(StockIntraday.code) + .group_by(StockIntraday.code) + .having(func.max(StockIntraday.datetime) >= func.date_sub(func.now(), interval=7 * 24 * 3600)) + ) + return {row[0] for row in result} + except Exception: + return set() + finally: + session.close() + + +def _code_to_sina(code: str) -> str: + if code.startswith(("6", "9")): + return f"sh{code}" + return f"sz{code}" + + +def fetch_intraday(): + """抓取1分钟分时行情（新浪接口，返回近5个交易日数据）""" + cfg = get_fetch_config() + delay = cfg.get("delay", 0.2) + retry = cfg.get("retry", 5) + + codes = _get_stock_codes() + if not codes: + print(" 无股票列表，请先运行 --stock-list", flush=True) + return + + existing = _get_existing_codes() + if existing: + print(f" 已有 {len(existing)} 只股票的分时数据，跳过...", flush=True) + codes = [c for c in codes if c not in existing] + + total = len(codes) + if total == 0: + print(" 所有股票分时数据已存在，无需抓取", flush=True) + return + + success = 0 + fail = 0 + consecutive_fail = 0 + t_start = time.time() + + print(f"正在抓取分时行情（1分钟线），需抓取 {total} 只（跳过 {len(existing)} 只）...", flush=True) + + for i, code in enumerate(codes): + sina_code = _code_to_sina(code) + t0 = time.time() + + df = None + for attempt in range(retry): + try: + df = ak.stock_zh_a_minute(symbol=sina_code, period="1") + consecutive_fail = 0 + break + except Exception as e: + wait = 3 * (attempt + 1) + if attempt < retry - 1: + print(f" [{i+1}/{total}] {code} 第{attempt+1}次重试，等待{wait}秒...", flush=True) + time.sleep(wait) + else: + print(f" [{i+1}/{total}] {code} 抓取失败: {e}", flush=True) + fail += 1 + consecutive_fail += 1 + + if consecutive_fail >= 5: + print(f" 连续{consecutive_fail}只失败，暂停60秒...", flush=True) + time.sleep(60) + consecutive_fail = 0 + + if df is None or df.empty: + time.sleep(delay) + continue + + records = df.to_dict(orient="records") + rows = [] + for rec in records: + rows.append({ + "code": code, + "datetime": rec.get("day"), + "open": rec.get("open"), + "high": rec.get("high"), + "low": rec.get("low"), + "close": rec.get("close"), + "volume": rec.get("volume"), + "amount": rec.get("amount"), + }) + + try: + batch_upsert(StockIntraday, rows, ["code", "datetime"]) + success += 1 + except Exception as e: + print(f" [{i+1}/{total}] {code} 写入失败: {e}", flush=True) + fail += 1 + + total_elapsed = time.time() - t_start + avg = total_elapsed / (i + 1) + eta = avg * (total - i - 1) + + print(f" [{i+1}/{total}] {code} 成功:{success} 失败:{fail} " + f"已用时:{total_elapsed:.0f}s 预计剩余:{eta:.0f}s", flush=True) + + time.sleep(delay) + + total_time = time.time() - t_start + print(f" 分时行情抓取完成，成功:{success} 失败:{fail} 总耗时:{total_time:.1f}s", flush=True) diff --git a/src/fetchers/money_flow.py b/src/fetchers/money_flow.py new file mode 100644 index 0000000..5af289c --- /dev/null +++ b/src/fetchers/money_flow.py @@ -0,0 +1,125 @@ +"""个股资金流向抓取模块 — 使用东方财富数据源 + +通过 AKShare 的 stock_individual_fund_flow 接口获取每只股票的历史资金流向， +包含主力/超大/大/中/小单的净流入金额和占比。 + +字段名使用多候选匹配（如 ["收盘价", "close"]）， +因为 AKShare 不同版本返回的列名可能为中文或英文。 +""" + +import time +from datetime import datetime +import akshare as ak +from src.config import get_fetch_config +from src.db import StockInfo, StockMoneyFlow, batch_upsert, get_session +from sqlalchemy import select + + +def _get_stock_codes() -> list[str]: + session = get_session() + try: + result = session.execute(select(StockInfo.code)) + return [row[0] for row in result] + finally: + session.close() + + +def fetch_money_flow(market: str = "sh"): + """抓取个股资金流向数据 + market: "sh" 或 "sz" + """ + cfg = get_fetch_config() + delay = cfg.get("delay", 1.0) + retry = cfg.get("retry", 5) + + codes = _get_stock_codes() + if not codes: + print(" 无股票列表，请先运行 --stock-list", flush=True) + return + + total = len(codes) + success = 0 + fail = 0 + consecutive_fail = 0 + + print(f"正在抓取资金流向数据，共 {total} 只股票...", flush=True) + + for i, code in enumerate(codes): + df = None + for attempt in range(retry): + try: + df = ak.stock_individual_fund_flow(stock=code, market=market) + consecutive_fail = 0 + break + except Exception as e: + wait = 3 * (attempt + 1) + if attempt < retry - 1: + print(f" [{i+1}/{total}] {code} 第{attempt+1}次重试，等待{wait}秒...", flush=True) + time.sleep(wait) + else: + print(f" [{i+1}/{total}] {code} 抓取失败: {e}", flush=True) + fail += 1 + consecutive_fail += 1 + + if consecutive_fail >= 5: + print(f" 连续{consecutive_fail}只失败，暂停60秒...", flush=True) + time.sleep(60) + consecutive_fail = 0 + + if df is None or df.empty: + time.sleep(delay) + continue + + rows = [] + for _, row in df.iterrows(): + date_val = row.get("日期", row.get("date", "")) + try: + rows.append({ + "code": code, + "date": date_val, + "close": _safe_float(row, ["收盘价", "close"]), + "pct_change": _safe_float(row, ["涨跌幅", "pct_change"]), + "main_net_inflow": _safe_float(row, ["主力净流入-净额", "main_net_inflow"]), + "main_net_pct": _safe_float(row, ["主力净流入-净占比", "main_net_pct"]), + "huge_net_inflow": _safe_float(row, ["超大盘净流入-净额", "huge_net_inflow"]), + "huge_net_pct": _safe_float(row, ["超大盘净流入-净占比", "huge_net_pct"]), + "big_net_inflow": _safe_float(row, ["大盘净流入-净额", "big_net_inflow"]), + "big_net_pct": _safe_float(row, ["大盘净流入-净占比", "big_net_pct"]), + "mid_net_inflow": _safe_float(row, ["中盘净流入-净额", "mid_net_inflow"]), + "mid_net_pct": _safe_float(row, ["中盘净流入-净占比", "mid_net_pct"]), + "small_net_inflow": _safe_float(row, ["小盘净流入-净额", "small_net_inflow"]), + "small_net_pct": _safe_float(row, ["小盘净流入-净占比", "small_net_pct"]), + }) + except Exception: + continue + + try: + batch_upsert(StockMoneyFlow, rows, ["code", "date"]) + success += 1 + except Exception as e: + print(f" [{i+1}/{total}] {code} 写入失败: {e}", flush=True) + fail += 1 + + if (i + 1) % 50 == 0: + print(f" [{i+1}/{total}] 进度... 成功:{success} 失败:{fail}", flush=True) + else: + print(f" [{i+1}/{total}] {code} OK", flush=True) + + time.sleep(delay) + + print(f" 资金流向抓取完成，成功:{success} 失败:{fail}", flush=True) + + +def _safe_float(row, keys: list[str]): + """从 DataFrame 行中按多个候选列名提取浮点值 + + AKShare 接口返回的列名可能在中文/英文之间变化， + 因此传入多个候选列名依次尝试。 + """ + val = row.get(key) + if val is not None: + try: + return float(val) + except (ValueError, TypeError): + return None + return None diff --git a/src/fetchers/stock_list.py b/src/fetchers/stock_list.py new file mode 100644 index 0000000..2338316 --- /dev/null +++ b/src/fetchers/stock_list.py @@ -0,0 +1,102 @@ +"""股票列表抓取模块 — 分别从沪深北三个交易所获取股票基本信息 + +AKShare 对三个交易所有独立的接口，返回字段名不一致： + - 沪市: 证券代码、证券简称、上市日期 + - 深市: A股代码、A股简称、A股上市日期 + - 北交所: 证券代码、证券简称、上市日期 +需要分别处理字段映射，最后按 code 去重合并。 +""" + +import time +import akshare as ak +from src.config import get_fetch_config +from src.db import StockInfo, batch_upsert + + +def fetch_stock_list(): + """抓取A股股票列表（含上市日期）""" + cfg = get_fetch_config() + retry = cfg.get("retry", 3) + + print("正在抓取A股股票列表（含上市日期）...", flush=True) + + rows = [] + + # 沪市 + print(" 抓取沪市...", flush=True) + for attempt in range(retry): + try: + df = ak.stock_info_sh_name_code(symbol="主板A股") + for _, r in df.iterrows(): + rows.append({ + "code": str(r["证券代码"]), + "name": r["证券简称"], + "ipo_date": _safe_date(r.get("上市日期")), + }) + print(f" 沪市 {len(df)} 只", flush=True) + break + except Exception as e: + if attempt < retry - 1: + time.sleep(2) + else: + print(f" 沪市抓取失败: {e}", flush=True) + + # 深市 + print(" 抓取深市...", flush=True) + for attempt in range(retry): + try: + df = ak.stock_info_sz_name_code() + for _, r in df.iterrows(): + rows.append({ + "code": str(r["A股代码"]), + "name": r["A股简称"], + "ipo_date": _safe_date(r.get("A股上市日期")), + }) + print(f" 深市 {len(df)} 只", flush=True) + break + except Exception as e: + if attempt < retry - 1: + time.sleep(2) + else: + print(f" 深市抓取失败: {e}", flush=True) + + # 北交所 + print(" 抓取北交所...", flush=True) + for attempt in range(retry): + try: + df = ak.stock_info_bj_name_code() + for _, r in df.iterrows(): + rows.append({ + "code": str(r["证券代码"]), + "name": r["证券简称"], + "ipo_date": _safe_date(r.get("上市日期")), + }) + print(f" 北交所 {len(df)} 只", flush=True) + break + except Exception as e: + if attempt < retry - 1: + time.sleep(2) + else: + print(f" 北交所抓取失败: {e}", flush=True) + + # 三个交易所可能有重叠代码（理论上不会），以 code 为主键去重 + seen = set() + unique = [] + for r in rows: + if r["code"] not in seen: + seen.add(r["code"]) + unique.append(r) + + batch_upsert(StockInfo, unique, ["code"]) + print(f" 股票列表抓取完成，共 {len(unique)} 只股票", flush=True) + return unique + + +def _safe_date(val): + """安全转换日期值，处理 pandas 的 NaT（Not a Time）等特殊值""" + if val is None: + return None + s = str(val).strip() + if not s or s == "None" or s == "NaT": + return None + return s[:10] diff --git a/src/fetchers/trading_day.py b/src/fetchers/trading_day.py new file mode 100644 index 0000000..7f0d253 --- /dev/null +++ b/src/fetchers/trading_day.py @@ -0,0 +1,70 @@ +"""交易日历模块 — 提供三级回退策略获取交易日列表 + +1. 本地 trading_day 表（最快，之前已缓存） +2. AKShare 新浪交易日历接口（从交易所获取官方日历） +3. 从 stock_daily 表已有数据推断（最后手段） + +交易日历是日线数据完整性判断的关键依据： +股票在某个日期范围内的记录数必须等于交易日数才算完整。 +""" + +import time +import akshare as ak +from src.db import TradingDay, batch_upsert, get_session +from sqlalchemy import select, func + + +def get_trading_days(start_date: str, end_date: str) -> list[str]: + """获取指定范围内的交易日列表，本地表优先，缺失则从数据源拉取""" + session = get_session() + try: + result = session.execute( + select(TradingDay.date) + .where(TradingDay.date >= start_date) + .where(TradingDay.date <= end_date) + .order_by(TradingDay.date) + ) + days = [str(row[0]) for row in result] + if days: + return days + finally: + session.close() + + # 本地无数据，从 AKShare 拉取交易所交易日历 + print(f" 正在从交易所获取交易日历 {start_date} ~ {end_date}...", flush=True) + try: + df = ak.tool_trade_date_hist_sina() + # AKShare 返回的 trade_date 列是 datetime.date 对象，不能直接和字符串比较 + # 必须先转为字符串再做范围过滤 + sd = f"{start_date[:4]}-{start_date[4:6]}-{start_date[6:8]}" + ed = f"{end_date[:4]}-{end_date[4:6]}-{end_date[6:8]}" + df["date_str"] = df["trade_date"].astype(str) + mask = (df["date_str"] >= sd) & (df["date_str"] <= ed) + filtered = df.loc[mask] + rows = [{"date": row["date_str"]} for _, row in filtered.iterrows()] + + if rows: + batch_upsert(TradingDay, rows, ["date"]) + print(f" 交易日历已保存，{len(rows)} 个交易日", flush=True) + return [r["date"] for r in rows] + except Exception as e: + print(f" 获取交易日历失败: {e}，将从已有行情数据推断", flush=True) + + # 回退：从 stock_daily 推断 + return _infer_from_daily(start_date, end_date) + + +def _infer_from_daily(start_date: str, end_date: str) -> list[str]: + """从 stock_daily 表推断交易日""" + from src.db import StockDaily + session = get_session() + try: + result = session.execute( + select(func.distinct(StockDaily.date)) + .where(StockDaily.date >= start_date) + .where(StockDaily.date <= end_date) + .order_by(StockDaily.date) + ) + return [str(row[0]) for row in result] + finally: + session.close() diff --git a/src/main.py b/src/main.py new file mode 100644 index 0000000..d6183d6 --- /dev/null +++ b/src/main.py @@ -0,0 +1,75 @@ +"""A股数据抓取工具主入口 + +用法示例： + python -m src.main --stock-list # 先抓取股票列表 + python -m src.main --daily --start-date 20260501 --end-date 20260508 + python -m src.main --all # 全量抓取 + python -m src.main --financial --symbol 000001 # 单只股票财务数据 +""" + +import argparse +from datetime import datetime + +from src.config import load_config +from src.db import init_db + + +def main(): + parser = argparse.ArgumentParser(description="A股数据抓取工具") + parser.add_argument("--stock-list", action="store_true", help="抓取股票列表") + parser.add_argument("--daily", action="store_true", help="抓取日线行情") + parser.add_argument("--financial", action="store_true", help="抓取财务报表") + parser.add_argument("--money-flow", action="store_true", help="抓取资金流向") + parser.add_argument("--dragon-tiger", action="store_true", help="抓取龙虎榜") + parser.add_argument("--dividend", action="store_true", help="抓取分红送转") + parser.add_argument("--intraday", action="store_true", help="抓取分时行情（1分钟线）") + parser.add_argument("--all", action="store_true", help="全量抓取所有数据") + parser.add_argument("--start-date", type=str, help="开始日期 YYYYMMDD") + parser.add_argument("--end-date", type=str, help="结束日期 YYYYMMDD") + parser.add_argument("--symbol", type=str, help="指定单只股票代码") + + args = parser.parse_args() + + if not any([args.stock_list, args.daily, args.financial, args.money_flow, + args.dragon_tiger, args.dividend, args.intraday, args.all]): + parser.print_help() + return + + # 加载配置并初始化数据库（自动建表） + load_config() + init_db() + + # 按需延迟导入各抓取模块，避免加载全部依赖 + if args.all or args.stock_list: + from src.fetchers.stock_list import fetch_stock_list + fetch_stock_list() + + if args.all or args.daily: + from src.fetchers.daily import fetch_daily + fetch_daily(start_date=args.start_date, end_date=args.end_date) + + if args.all or args.financial: + from src.fetchers.financial import fetch_financial + fetch_financial(symbol=args.symbol) + + if args.all or args.money_flow: + from src.fetchers.money_flow import fetch_money_flow + fetch_money_flow() + + if args.all or args.dragon_tiger: + from src.fetchers.dragon_tiger import fetch_dragon_tiger + fetch_dragon_tiger(start_date=args.start_date, end_date=args.end_date) + + if args.all or args.dividend: + from src.fetchers.dividend import fetch_dividend + fetch_dividend(symbol=args.symbol) + + if args.all or args.intraday: + from src.fetchers.intraday import fetch_intraday + fetch_intraday() + + print("全部任务完成", flush=True) + + +if __name__ == "__main__": + main()