zoe/ashare-data
SHA256
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

init

Browse Source
This commit is contained in:
曾志威
2026-05-10 12:11:49 +08:00
parent feef7b63d0
commit e2c3b41b27
16 changed files with 1028 additions and 1117 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+108 -104
View File
@@ -6,12 +6,15 @@ A股数据抓取工具，使用 [AKShare](https://github.com/akfamily/akshare)
| 数据类型 | 说明 | 数据来源 |
|---------|------|---------|
| 股票列表 | A股全部股票代码和名称 | 东方财富 |
| 日线行情 | 开盘价、收盘价、最高价、最低价、成交量、成交额、涨跌幅等（前复权） | 东方财富 |
| 财务报表 | 利润表、资产负债表、现金流量表 | 新浪财经 |
| 资金流向 | 主力/超大/大/中/小单净流入及占比 | 东方财富 |
| 龙虎榜 | 上榜股票、买入卖出金额、上榜原因 | 东方财富 |
| 分红送转 | 每10股送转、派息、股息率等 | 巨潮资讯 |
| 股票列表 | 沪深A股代码、名称、上市日期 | BaoStock |
| 交易日历 | 1990年至今的交易日列表 | BaoStock |
| 日线行情 | 开高低收、成交量/额、振幅、涨跌幅、换手率（前复权） | BaoStock |
| 季频财务指标 | 盈利能力、偿债能力、现金流（最近8个季度） | BaoStock |
| 分红送转 | 每10股送转、派息、除权除息日（最近10年） | BaoStock |
| 分时行情 | 5/15/30/60分钟K线（开高低收、成交量/额） | BaoStock |
| 行业+地域分类 | 证监会行业分类 + 省份 | BaoStock + 东方财富 |
**已知限制**：BaoStock 不含北交所（920xxx）股票。
## 快速开始
@@ -23,11 +26,7 @@ A股数据抓取工具，使用 [AKShare](https://github.com/akfamily/akshare)
### 2. 安装依赖
```bash
python -m venv .venv
.venv\Scripts\activate # Windows
# source .venv/bin/activate # Linux/Mac
pip install -r requirements.txt
pip install baostock pymysql sqlalchemy pyyaml pandas
```
### 3. 配置数据库
@@ -38,7 +37,7 @@ pip install -r requirements.txt
cp config.example.yaml config.yaml
```
编辑 `config.yaml`，填写你的 MySQL 连接信息：
编辑 `config.yaml`：
```yaml
mysql:
@@ -50,8 +49,9 @@ mysql:
charset: "utf8mb4"
fetch:
delay: 0.5 # 请求间隔(秒)，防止被限流
retry: 3 # 失败重试次数
delay: 0.1 # 请求间隔(秒)
retry: 2 # 失败重试次数
workers: 1 # 并发线程数
```
确保 MySQL 中已创建对应数据库：
@@ -63,60 +63,61 @@ CREATE DATABASE ashare CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
### 4. 运行
```bash
# 先抓取股票列表（其他模块依赖此数据）
python -m src.main --stock-list
# 1. 先抓取股票列表（其他模块依赖此数据）
python -m src.main --stock-info
# 抓取最近30天的日线行情
# 2. 抓取交易日历
python -m src.main --trading-day --start-date 19901219 --end-date 20261231
# 3. 抓取最近30天的日线行情
python -m src.main --daily
# 指定日期范围抓取日线
python -m src.main --daily --start-date 20260430 --end-date 20260508
python -m src.main --daily --start-date 19901201 --end-date 20260508
# 抓取财务报表（全部股票）
# 抓取财务指标（全部股票，最近8个季度）
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
# 抓取分红送转
# 抓取分红送转（全部股票，最近10年）
python -m src.main --dividend
# 全量抓取所有数据
python -m src.main --all
# 抓取单只股票的分红
python -m src.main --dividend --symbol 000001
# 抓取分钟K线（默认5分钟）
python -m src.main --intraday
# 抓取全部频率分钟K线
python -m src.main --intraday --freq all --start-date 20260508
# 抓取行业+地域分类
python -m src.main --sector
```
### 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 全量抓取以上所有数据
--stock-info 抓取A股股票列表
--trading-day 抓取交易日历
--daily 抓取日线行情
--financial 抓取季频财务指标
--dividend 抓取分红送转数据
--intraday 抓取分钟K线行情
--freq K线频率: 5/15/30/60/all（默认5）
--sector 抓取行业+地域分类
--industry-only 仅抓取行业分类
--region-only 仅抓取地域分类
日期过滤（仅对日线行情和龙虎榜生效）:
--start-date 开始日期，格式 YYYYMMDD，默认30天前
--end-date 结束日期，格式 YYYYMMDD，默认今天
日期过滤（对日线行情、交易日历、分钟K线生效）:
--start-date 开始日期，格式 YYYYMMDD
--end-date 结束日期，格式 YYYYMMDD
股票过滤（仅对财务报表和分红送转生效）:
--symbol 指定单只股票代码，如 000001，默认全部股票
股票过滤（对财务指标、分红送转、分钟K线生效）:
--symbol 指定单只股票代码，如 000001，默认全部股票
```
## 数据库表结构
@@ -127,6 +128,7 @@ usage: main.py [-h] [--stock-list] [--daily] [--financial] [--money-flow]
|------|------|------|
| code | VARCHAR(10) PK | 股票代码 |
| name | VARCHAR(50) | 股票名称 |
| ipo_date | DATE | 上市日期 |
### stock_daily — 日线行情（前复权）
@@ -147,73 +149,36 @@ usage: main.py [-h] [--stock-list] [--daily] [--financial] [--money-flow]
联合主键：`(code, date)`
### stock_financial_income — 利润表
### stock_financial_income — 季频盈利能力
| 字段 | 类型 | 说明 |
|------|------|------|
| code | VARCHAR(10) | 股票代码 |
| report_date | VARCHAR(20) | 报告期 |
| data | TEXT | JSON格式利润表数据 |
| data | TEXT | JSON格式数据（ROE、净利率、毛利率等） |
联合主键：`(code, report_date)`
### stock_financial_balance — 资产负债表
### stock_financial_balance — 季频偿债能力
| 字段 | 类型 | 说明 |
|------|------|------|
| code | VARCHAR(10) | 股票代码 |
| report_date | VARCHAR(20) | 报告期 |
| data | TEXT | JSON格式资产负债表数据 |
| data | TEXT | JSON格式数据（流动比率、资产负债率等） |
联合主键：`(code, report_date)`
### stock_financial_cashflow — 现金流量表
### stock_financial_cashflow — 季频现金流
| 字段 | 类型 | 说明 |
|------|------|------|
| code | VARCHAR(10) | 股票代码 |
| report_date | VARCHAR(20) | 报告期 |
| data | TEXT | JSON格式现金流量表数据 |
| 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 — 分红送转
| 字段 | 类型 | 说明 |
@@ -230,34 +195,73 @@ usage: main.py [-h] [--stock-list] [--daily] [--financial] [--money-flow]
联合主键：`(code, report_date)`
### trading_day — 交易日历
| 字段 | 类型 | 说明 |
|------|------|------|
| date | DATE UQ | 交易日期 |
### stock_no_data — 停牌/无数据记录
| 字段 | 类型 | 说明 |
|------|------|------|
| code | VARCHAR(10) | 股票代码 |
| date | DATE | 停牌/无数据日期 |
联合主键：`(code, date)`
### stock_min5 / stock_min15 / stock_min30 / stock_min60 — 分钟K线
| 字段 | 类型 | 说明 |
|------|------|------|
| code | VARCHAR(10) | 股票代码 |
| datetime | DATETIME | 时间 |
| open | FLOAT | 开盘价 |
| high | FLOAT | 最高价 |
| low | FLOAT | 最低价 |
| close | FLOAT | 收盘价 |
| volume | FLOAT | 成交量 |
| amount | FLOAT | 成交额 |
联合主键：`(code, datetime)`
### stock_sector — 行业+地域分类
| 字段 | 类型 | 说明 |
|------|------|------|
| code | VARCHAR(10) PK | 股票代码 |
| industry | VARCHAR(100) | 证监会行业分类 |
| region | VARCHAR(20) | 省份/地域 |
## 项目结构
```
ashare-data/
a股数据看板/
├── config.yaml # MySQL连接配置（.gitignore）
├── config.example.yaml # 配置文件模板
├── config.yaml # 实际配置（含密码，已加入.gitignore）
├── pyproject.toml # 项目元数据
├── requirements.txt # Python依赖
├── src/
│ ├── __init__.py
│ ├── config.py # 配置读取模块
│ ├── baostock_conn.py # BaoStock连接管理（login/logout/线程锁）
│ ├── db.py # 数据库模型与连接管理
│ ├── main.py # 命令行入口
│ └── fetchers/
│ ├── __init__.py
│ ├── stock_list.py # 股票列表抓取
│ ├── daily.py # 日线行情抓取
│ ├── financial.py # 财务报表抓取
│ ├── money_flow.py # 资金流向抓取
│ ├── dragon_tiger.py # 龙虎榜抓取
│ └── dividend.py # 分红送转抓取
│ ├── stock_list.py # 股票列表
│ ├── trading_day.py # 交易日历
│ ├── daily.py # 日线行情
│ ├── financial.py # 季频财务指标
│ ├── dividend.py # 分红送转
│ ├── intraday.py # 分钟K线（5/15/30/60）
│ └── sector.py # 行业+地域分类
└── README.md
```
## 设计说明
- **单数据源**：统一使用 BaoStock，无需 AKShare 等额外依赖
- **线程安全**：BaoStock 查询通过全局锁串行化，避免数据错乱
- **去重写入**：所有表使用 `INSERT ON DUPLICATE KEY UPDATE`（upsert），重复执行不会产生重复数据
- **自动重试**：网络请求失败自动重试（默认3次），单只股票失败不影响整体
- **限速保护**：请求间自动延迟（默认0.5秒），防止被数据源限流
- **增量更新**：日线行情和龙虎榜支持通过 `--start-date` / `--end-date` 指定日期范围
- **懒加载导入**：各fetcher模块按需导入，未使用的模块不会加载
- **增量更新**：日线行情支持通过 `--start-date` / `--end-date` 指定日期范围，已完整的数据自动跳过
- **停牌标记**：缺失交易日自动标记为停牌，避免重复抓取
config.yaml
+1 -1
View File
@@ -11,4 +11,4 @@ mysql:
fetch:
delay: 0.1 # 每次请求间隔（秒）
retry: 2 # 失败重试次数
workers: 3 # 并发线程数；与数据源数量一致（BaoStock+新浪+腾讯）
workers: 1 # 并发线程数；与数据源数量一致（BaoStock+新浪+腾讯）
pyproject.toml
+1
View File
@@ -9,6 +9,7 @@ dependencies = [
"sqlalchemy>=2.0",
"pyyaml",
"pandas",
"requests",
]
[project.scripts]
src/baostock_conn.py
+53
View File
@@ -0,0 +1,53 @@
"""BaoStock 连接管理器 — 集中管理 login/logout/线程锁/代码格式转换
BaoStock 的 query_xxx() 非线程安全，所有查询需通过同一把锁串行化。
"""
import threading
from contextlib import contextmanager
import baostock as bs
_lock = threading.Lock()
_logged_in = False
def bs_login():
"""全局只 login 一次"""
global _logged_in
with _lock:
if not _logged_in:
bs.login()
_logged_in = True
def bs_logout():
"""程序退出时调用"""
global _logged_in
with _lock:
if _logged_in:
bs.logout()
_logged_in = False
@contextmanager
def bs_query(query_fn, *args, **kwargs):
"""加锁执行 BaoStock 查询，yield ResultData
用法:
with bs_query(bs.query_history_k_data_plus, code, fields, ...) as rs:
while rs.next():
row = rs.get_row_data()
"""
bs_login()
with _lock:
rs = query_fn(*args, **kwargs)
yield rs
def code_to_bs(code: str) -> str:
"""纯数字代码转 BaoStock 格式: '600000' → 'sh.600000'"""
if code.startswith("920"):
return None # 北交所不支持
if code.startswith(("6", "9")):
return f"sh.{code}"
return f"sz.{code}"
src/db.py
+125 -76
View File
@@ -1,15 +1,15 @@
"""数据库模型定义与连接管理
数据源：BaoStock（唯一数据源）
表结构概览：
- stock_info: 股票基本信息（含上市日期，用于跳过未上市股票）
- stock_daily: 日线行情（多源抓取：BaoStock/新浪/腾讯）
- stock_financial_income/balance/cashflow: 三大财务报表（JSON存储）
- stock_money_flow: 个股资金流向
- stock_dragon_tiger: 龙虎榜
- stock_daily: 日线行情（BaoStock，含振幅/涨跌幅/换手率）
- stock_financial_income/balance/cashflow: 季频财务指标（JSON存储）
- stock_dividend: 分红送转
- stock_intraday: 1分钟分时行情
- trading_day: 交易日历（用于判断数据完整性）
- stock_no_data: 无数据/停牌记录（避免重复抓取）
- stock_intraday: 5分钟分时行情
- market_breadth: 市场量能快照
"""
from sqlalchemy import (
@@ -32,7 +32,6 @@ class StockInfo(Base):
code = Column(String(10), primary_key=True, comment="股票代码")
name = Column(String(50), comment="股票名称")
# 上市日期用于在抓取历史数据时跳过当时尚未上市的股票
ipo_date = Column(Date, comment="上市日期")
@@ -46,7 +45,6 @@ class StockDaily(Base):
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="收盘价")
@@ -60,7 +58,7 @@ class StockDaily(Base):
turnover_rate = Column(Float, comment="换手率%")
# ── 利润表 ──
# ── 季频盈利能力 ──
class FinancialIncome(Base):
__tablename__ = "stock_financial_income"
__table_args__ = (
@@ -70,10 +68,10 @@ class FinancialIncome(Base):
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格式利润表数据")
data = Column(Text, comment="JSON格式盈利能力数据")
# ── 资产负债表 ──
# ── 季频营运能力 ──
class FinancialBalance(Base):
__tablename__ = "stock_financial_balance"
__table_args__ = (
@@ -83,10 +81,10 @@ class FinancialBalance(Base):
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格式资产负债表数据")
data = Column(Text, comment="JSON格式营运能力数据")
# ── 现金流量表 ──
# ── 季频现金流 ──
class FinancialCashflow(Base):
__tablename__ = "stock_financial_cashflow"
__table_args__ = (
@@ -96,52 +94,7 @@ class FinancialCashflow(Base):
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="净额")
data = Column(Text, comment="JSON格式现金流数据")
# ── 分红送转 ──
@@ -187,12 +140,12 @@ class StockNoData(Base):
created_at = Column(DateTime, server_default=func.now(), comment="记录时间")
# ── 分时行情（1分钟线） ──
class StockIntraday(Base):
__tablename__ = "stock_intraday"
# ── 5分钟K线 ──
class StockMin5(Base):
__tablename__ = "stock_min5"
__table_args__ = (
UniqueConstraint("code", "datetime", name="uq_intraday_code_dt"),
Index("ix_intraday_date", "datetime"),
UniqueConstraint("code", "datetime", name="uq_min5_code_dt"),
Index("ix_min5_datetime", "datetime"),
)
id = Column(Integer, primary_key=True, autoincrement=True)
@@ -206,6 +159,89 @@ class StockIntraday(Base):
amount = Column(Float, comment="成交额")
# ── 15分钟K线 ──
class StockMin15(Base):
__tablename__ = "stock_min15"
__table_args__ = (
UniqueConstraint("code", "datetime", name="uq_min15_code_dt"),
Index("ix_min15_datetime", "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="成交额")
# ── 30分钟K线 ──
class StockMin30(Base):
__tablename__ = "stock_min30"
__table_args__ = (
UniqueConstraint("code", "datetime", name="uq_min30_code_dt"),
Index("ix_min30_datetime", "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="成交额")
# ── 60分钟K线 ──
class StockMin60(Base):
__tablename__ = "stock_min60"
__table_args__ = (
UniqueConstraint("code", "datetime", name="uq_min60_code_dt"),
Index("ix_min60_datetime", "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="成交额")
# ── 每日涨跌停统计 ──
class MarketDaily(Base):
__tablename__ = "market_daily"
__table_args__ = (
UniqueConstraint("date", name="uq_market_daily_date"),
)
date = Column(Date, primary_key=True, comment="交易日期")
limit_up_10 = Column(Integer, comment="涨停数（10%主板）")
limit_up_20 = Column(Integer, comment="涨停数（20%科创/创业板）")
limit_down_10 = Column(Integer, comment="跌停数（10%主板）")
limit_down_20 = Column(Integer, comment="跌停数（20%科创/创业板）")
# ── 行业+地域分类 ──
class StockSector(Base):
__tablename__ = "stock_sector"
__table_args__ = (
UniqueConstraint("code", name="uq_sector_code"),
)
code = Column(String(10), primary_key=True, comment="股票代码")
industry = Column(String(100), comment="行业分类")
region = Column(String(20), comment="省份/地域")
# ── 数据库连接管理 ──
_engine = None
_SessionFactory = None
@@ -228,29 +264,43 @@ def get_session() -> Session:
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"))
try:
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）")
except Exception:
pass
# 自动迁移：旧版 stock_sector 使用 update_date 列，新版改为 region
try:
with engine.connect() as conn:
result = conn.execute(text("SHOW COLUMNS FROM stock_sector LIKE 'update_date'"))
if result.fetchone():
conn.execute(text("DROP TABLE stock_sector"))
conn.commit()
print(" stock_sector 表结构已升级（新增 region 列）")
except Exception:
pass
# 自动迁移：旧版 stock_intraday 单表 → 四张分表
try:
with engine.connect() as conn:
conn.execute(text("DROP TABLE IF EXISTS stock_intraday"))
conn.commit()
print(" stock_no_data 表结构已升级（date_range → date）")
except Exception:
pass
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"]），
这些列在冲突时不更新，其余列使用新值覆盖。
"""
"""MySQL批量upsert：INSERT ON DUPLICATE KEY UPDATE"""
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]
@@ -263,7 +313,6 @@ def batch_upsert(model_cls: type[Base], rows: list[dict], index_columns: list[st
if update_dict:
stmt = stmt.on_duplicate_key_update(**update_dict)
else:
# 无可更新列时（如 TradingDay 只有唯一键列），用唯一键本身做 no-op 更新
stmt = stmt.on_duplicate_key_update(**{index_columns[0]: stmt.inserted[index_columns[0]]})
session.execute(stmt)
session.commit()
src/fetchers/daily.py
+67 -260
View File
@@ -1,10 +1,6 @@
"""日线行情抓取模块 — 核心模块，采用三数据源轮询 + 自动降级架构
数据源优先级（按速度排序）：
1. BaoStock: 速度极快（~0.04s/只），但不支持北交所（920xxx），线程不安全需加锁
2. 新浪: 支持全部交易所（含北交所 bj 前缀），返回 JSONP 需解析
3. 腾讯: 不支持北交所，返回标准 JSON
"""日线行情抓取模块 — 使用 BaoStock
单数据源架构，代码大幅简化。
跳过策略（停牌天按天记录）：
- 数据完整 = 行情记录数 + 已标记停牌天数 >= 交易日总数
- 未上市股票（ipo_date > 查询结束日期）
@@ -14,43 +10,21 @@
import json
import re
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timedelta
import requests
import baostock as bs
from src.baostock_conn import bs_query, code_to_bs, bs_login
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 _clean(val):
"""将空字符串、无效值转为 None"""
if val is None:
return None
if isinstance(val, str) and val.strip() == "":
return None
return val
def _get_stock_codes() -> list[str]:
@@ -63,9 +37,7 @@ def _get_stock_codes() -> list[str]:
def _get_not_listed(end_date: str) -> set[str]:
"""查询在end_date之后上市的股票（未上市，需跳过）
例如抓取 20260501~20260508 的数据时，5月10日上市的股票应被跳过
"""
"""查询在end_date之后上市的股票（未上市，需跳过）"""
session = get_session()
try:
result = session.execute(
@@ -77,17 +49,12 @@ def _get_not_listed(end_date: str) -> set[str]:
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)
@@ -96,7 +63,6 @@ def _get_complete_codes(start_date: str, end_date: str, trading_days: list[str])
)
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)
@@ -115,191 +81,64 @@ def _get_complete_codes(start_date: str, end_date: str, trading_days: list[str])
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"):
"""从 BaoStock 获取日线行情，含振幅/涨跌幅/换手率"""
bs_code = code_to_bs(code)
if not bs_code:
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",
)
with bs_query(
bs.query_history_k_data_plus,
bs_code,
"date,open,high,low,close,volume,amount,preclose,pctChg,turn",
start_date=sd, end_date=ed, frequency="d", adjustflag="2",
) as rs:
rows = []
while (rs.error_code == "0") and rs.next():
r = rs.get_row_data()
preclose = _clean(r[7])
pct_chg = _clean(r[8])
turn = _clean(r[9])
close = _clean(r[4])
high = _clean(r[2])
low = _clean(r[3])
amp = None
if high is not None and low is not None and preclose and float(preclose) != 0:
amp = round((float(high) - float(low)) / float(preclose) * 100, 2)
chg = None
if close is not None and preclose is not None:
try:
chg = round(float(close) - float(preclose), 3)
except (ValueError, TypeError):
pass
rows.append({
"code": code, "date": r[0],
"open": _clean(r[1]), "high": _clean(r[2]), "low": _clean(r[3]), "close": _clean(r[4]),
"open": _clean(r[1]), "high": high, "low": low, "close": close,
"volume": _clean(r[5]), "turnover": _clean(r[6]),
"amplitude": None, "pct_change": None, "change": None, "turnover_rate": None,
"amplitude": amp, "pct_change": _clean(pct_chg), "change": chg,
"turnover_rate": _clean(turn),
})
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", 3)
codes = _get_stock_codes()
if not codes:
print(" 无股票列表，请先运行 --stock-list", flush=True)
print(" 无股票列表，请先运行 --stock-info", flush=True)
return
if end_date is None:
@@ -307,21 +146,13 @@ def fetch_daily(start_date: str | None = None, end_date: str | None = None):
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
bs_login()
# 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
@@ -331,9 +162,6 @@ def fetch_daily(start_date: str | None = None, end_date: str | None = None):
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)
@@ -345,54 +173,33 @@ def fetch_daily(start_date: str | None = None, end_date: str | None = None):
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} 只）...", flush=True)
print(f"正在抓取日线行情 {start_date} ~ {end_date}，需抓取 {total} 只（跳过 {skipped} 只），"
f"{workers} 线程 × 3 源(BaoStock+新浪+腾讯)...", flush=True)
for i, code in enumerate(codes):
rows = _fetch_baostock(code, start_date, end_date)
if rows is not None:
try:
batch_upsert(StockDaily, rows, ["code", "date"])
success += 1
returned_dates = {row["date"] for row in 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:
# 无数据：标记所有交易日为停牌
_record_nodata_days(code, trading_days)
nodata_count += 1
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
total_elapsed = time.time() - t_start
avg = total_elapsed / (i + 1)
eta = avg * (total - i - 1)
print(f" [{i+1}/{total}] {code} 成功:{success} 失败:{fail} 停牌:{nodata_count} "
f"已用时:{total_elapsed:.0f}s 预计剩余:{eta:.0f}s", flush=True)
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)
time.sleep(delay)
total_time = time.time() - t_start
print(f" 日线行情抓取完成，成功:{success} 失败:{fail} 停牌:{nodata_count} 总耗时:{total_time:.1f}s", flush=True)
src/fetchers/dividend.py
+65 -90
View File
@@ -1,13 +1,12 @@
"""分红送转数据抓取模块 — 使用巨潮信息网（CNInfo）数据源
"""分红送转抓取模块 — 使用 BaoStock
通过 AKShare 的 stock_dividend_cninfo 接口按股票代码逐个查询，
返回该股票历史所有分红记录（含送股、转增、派息等）。
BaoStock query_dividend_data() 按年度查询分红记录。
迭代最近10年获取完整分红历史。
"""
import time
from datetime import datetime
import akshare as ak
from src.config import get_fetch_config
import baostock as bs
from src.baostock_conn import bs_query, code_to_bs
from src.db import StockInfo, StockDividend, batch_upsert, get_session
from sqlalchemy import select
@@ -21,100 +20,76 @@ def _get_stock_codes() -> list[str]:
session.close()
def fetch_dividend(symbol: str | None = None):
"""抓取分红送转数据"""
cfg = get_fetch_config()
delay = cfg.get("delay", 1.0)
retry = cfg.get("retry", 5)
def _fetch_dividend(code: str) -> list[dict]:
"""抓取单只股票最近10年的分红记录"""
bs_code = code_to_bs(code)
if not bs_code:
return []
current_year = datetime.now().year
rows = []
for year in range(current_year - 10, current_year + 1):
try:
with bs_query(bs.query_dividend_data, code=bs_code, year=str(year), yearType="report") as rs:
while rs.next():
r = rs.get_row_data()
# fields: code, dividPreNoticeDate, dividAgmPumDate, dividPlanAnnounceDate,
# dividPlanDate, dividRegistDate, dividOperateDate, dividPayDate,
# dividStockMarketDate, dividCashPsBeforeTax, dividCashPsAfterTax,
# dividStocksPs, dividCashStock, dividReserveToStockPs
operate_date = r[6] if len(r) > 6 else ""
cash_before_tax = r[9] if len(r) > 9 else ""
stock_ps = r[11] if len(r) > 11 else ""
reserve_ps = r[13] if len(r) > 13 else ""
if not operate_date and not cash_before_tax and not stock_ps:
continue
rows.append({
"code": code,
"name": "",
"report_date": str(year),
"dividend_date": operate_date or None,
"bonus_ratio": _to_float(stock_ps, scale=10),
"cash_div": _to_float(cash_before_tax, scale=10),
"convert_ratio": _to_float(reserve_ps, scale=10),
"ex_right_date": operate_date or None,
"dividend_yield": None,
})
except Exception:
continue
return rows
def _to_float(val, scale=1) -> float | None:
if not val or val == "0.000000":
return None
try:
return round(float(val) * scale, 4)
except (ValueError, TypeError):
return None
def fetch_dividend(symbol: str | None = None):
"""抓取分红送转数据
用法：python -m src.main --dividend [--symbol 000001]
"""
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)
success = 0
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:
rows = _fetch_dividend(code)
if rows:
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) % 100 == 0:
print(f" [{i+1}/{total}] 进度... 成功:{success}", flush=True)
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
print(f" 分红送转抓取完成，成功:{success}/{total}", flush=True)
src/fetchers/dragon_tiger.py
-69
View File
@@ -1,69 +0,0 @@
"""龙虎榜数据抓取模块 — 使用东方财富数据源
龙虎榜按日期范围查询，返回该期间内所有上榜股票的买卖详情。
数据量相对较小（每天几十到上百条），通常一次请求即可获取全部数据。
"""
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
src/fetchers/financial.py
+87 -108
View File
@@ -1,18 +1,17 @@
"""财务报表抓取模块 — 利润表、资产负债表、现金流量表
"""季频财务指标抓取模块 — 使用 BaoStock
使用 AKShare 的新浪财务数据接口，每个接口返回一个报告期对应的所有字段。
由于不同股票的字段名可能变化，采用 JSON 格式存储完整数据而非逐字段建列。
BaoStock 按季度查询财务数据：
- query_profit_data() 盈利能力
- query_balance_data() 偿债能力
- query_cash_flow_data() 现金流
防限流策略：
- 指数退避重试（3s、6s、9s...）
- 连续5只股票失败时暂停60秒（可能触发了限流）
数据以 JSON 格式存入 data 列（与现有表结构兼容）。
"""
import json
import time
from datetime import datetime
import akshare as ak
from src.config import get_fetch_config
import baostock as bs
from src.baostock_conn import bs_query, code_to_bs
from src.db import StockInfo, FinancialIncome, FinancialBalance, FinancialCashflow, batch_upsert, get_session
from sqlalchemy import select
@@ -26,112 +25,92 @@ def _get_stock_codes() -> list[str]:
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 _recent_quarters(n: int) -> list[tuple[int, int]]:
"""生成最近 n 个季度 [(year, quarter), ...]"""
now = datetime.now()
year, quarter = now.year, (now.month - 1) // 3 + 1
result = []
for _ in range(n):
result.append((year, quarter))
quarter -= 1
if quarter == 0:
quarter = 4
year -= 1
return result
def _parse_financial_df(code: str, df) -> list[dict]:
"""将财务报表 DataFrame 转为数据库行，所有字段以 JSON 格式存储
因为不同报表的字段名和数量差异很大，且可能随时间变化，
不适合逐字段建列。JSON 存储保留了原始数据的完整性。
"""
def _parse_resultset(code: str, rs, fields: list[str], year: int, quarter: int) -> list[dict]:
"""将 BaoStock ResultData 转为 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}
while rs.next():
r = rs.get_row_data()
# fields: code, pubDate, statDate, ...指标字段
stat_date = r[2] if len(r) > 2 else f"{year}-Q{quarter}"
data_dict = {}
for j, field in enumerate(fields):
if j < len(r):
val = r[j]
if isinstance(val, str) and val.strip() == "":
val = None
data_dict[field] = val
rows.append({
"code": code,
"report_date": report_date,
"report_date": stat_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
def fetch_financial(symbol: str | None = None):
"""抓取财务数据
用法：python -m src.main --financial [--symbol 000001]
默认抓取所有股票最近 8 个季度。
"""
if symbol:
codes = [symbol]
else:
codes = _get_stock_codes()
total = len(codes)
quarters = _recent_quarters(8)
print(f"正在抓取财务数据，共 {total} 只股票 × {len(quarters)} 个季度...", flush=True)
success = 0
fail = 0
for i, code in enumerate(codes):
bs_code = code_to_bs(code)
if not bs_code:
continue
try:
for year, quarter in quarters:
# 盈利能力
with bs_query(bs.query_profit_data, code=bs_code, year=year, quarter=quarter) as rs:
fields = rs.fields if rs.fields else []
rows = _parse_resultset(code, rs, fields, year, quarter)
if rows:
batch_upsert(FinancialIncome, rows, ["code", "report_date"])
# 偿债能力
with bs_query(bs.query_balance_data, code=bs_code, year=year, quarter=quarter) as rs:
fields = rs.fields if rs.fields else []
rows = _parse_resultset(code, rs, fields, year, quarter)
if rows:
batch_upsert(FinancialBalance, rows, ["code", "report_date"])
# 现金流
with bs_query(bs.query_cash_flow_data, code=bs_code, year=year, quarter=quarter) as rs:
fields = rs.fields if rs.fields else []
rows = _parse_resultset(code, rs, fields, year, quarter)
if rows:
batch_upsert(FinancialCashflow, 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 or i == 0:
print(f" [{i+1}/{total}] 进度... 成功:{success} 失败:{fail}", flush=True)
print(f" 财务数据抓取完成，成功:{success} 失败:{fail}", flush=True)
src/fetchers/intraday.py
+133 -89
View File
@@ -1,15 +1,35 @@
"""分时行情抓取模块（1分钟线）— 使用新浪数据源
"""分钟K线抓取模块 — 使用 BaoStock
注意：分时数据量非常大（每只股票每天约240条分钟记录），
目前暂未启用此模块的抓取任务。
BaoStock 支持 5/15/30/60 分钟K线，分别存入四张表。
注意：BaoStock 不支持 1 分钟线，且不含北交所。
用法：
python -m src.main --intraday --freq 5 # 5分钟K线
python -m src.main --intraday --freq all # 全部频率
python -m src.main --intraday --start-date 20260508 --end-date 20260509
python -m src.main --intraday --symbol 000001 --freq 30
"""
import time
from datetime import datetime
import akshare as ak
from datetime import datetime, timedelta
import baostock as bs
from src.baostock_conn import bs_query, code_to_bs, bs_login
from src.config import get_fetch_config
from src.db import StockInfo, StockIntraday, batch_upsert, get_session
from sqlalchemy import select, func
from src.db import (
StockInfo, StockMin5, StockMin15, StockMin30, StockMin60,
batch_upsert, get_session,
)
from sqlalchemy import select, func, distinct
VALID_FREQ = ("5", "15", "30", "60")
FREQ_MODEL = {
"5": StockMin5,
"15": StockMin15,
"30": StockMin30,
"60": StockMin60,
}
def _get_stock_codes() -> list[str]:
@@ -21,117 +41,141 @@ def _get_stock_codes() -> list[str]:
session.close()
def _get_existing_codes() -> set[str]:
"""查询已有分时数据的股票代码（最近5个交易日有数据的视为已有）
def _clean(val):
if val is None:
return None
if isinstance(val, str) and val.strip() == "":
return None
return val
分时数据时效性强，只保留近期数据即可，避免重复抓取。
def _parse_datetime(date_str: str, time_str: str) -> str | None:
"""将 BaoStock 返回的 date + time 解析为 datetime 字符串
time 格式: "20260508093500000" (17位) 或 "09:35:00" (8位)
"""
if len(time_str) == 17:
return f"{time_str[:4]}-{time_str[4:6]}-{time_str[6:8]} " \
f"{time_str[8:10]}:{time_str[10:12]}:{time_str[12:14]}"
return f"{date_str} {time_str}"
def _get_fetched_codes(model, sd: str, ed: str) -> set[str]:
"""查询已有数据的股票代码（在日期范围内有记录的）"""
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))
select(distinct(model.code))
.where(model.datetime >= sd)
.where(model.datetime <= ed + " 23:59:59")
)
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(start_date: str | None = None, end_date: str | None = None,
symbol: str | None = None, freq: str = "5"):
"""抓取分钟K线行情
Args:
start_date: 开始日期 YYYYMMDD，默认30天前
end_date: 结束日期 YYYYMMDD，默认今天
symbol: 单只股票代码，默认全部
freq: K线频率 5/15/30/60/all
"""
if freq == "all":
freqs = list(VALID_FREQ)
elif freq in VALID_FREQ:
freqs = [freq]
else:
print(f" 不支持的频率 {freq}，可选: {', '.join(VALID_FREQ)}, all", flush=True)
return
for f in freqs:
_fetch_one_freq(f, start_date, end_date, symbol)
def fetch_intraday():
"""抓取1分钟分时行情（新浪接口，返回近5个交易日数据）"""
def _fetch_one_freq(freq: str, start_date: str | None, end_date: str | None,
symbol: str | None):
"""抓取单个频率的分钟K线"""
cfg = get_fetch_config()
delay = cfg.get("delay", 0.2)
retry = cfg.get("retry", 5)
delay = cfg.get("delay", 0.1)
model = FREQ_MODEL[freq]
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")
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]}"
if symbol:
codes = [symbol]
else:
codes = _get_stock_codes()
if not codes:
print(" 无股票列表，请先运行 --stock-info", flush=True)
return
# 跳过已有数据的股票
fetched = set()
if not symbol:
fetched = _get_fetched_codes(model, sd, ed)
codes = [c for c in codes if c not in fetched]
codes = _get_stock_codes()
if not codes:
print(" 无股票列表，请先运行 --stock-list", flush=True)
print(f" {freq}分钟K线 {sd} ~ {ed} 数据已完整，跳过", 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]
bs_login()
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)
skip_msg = f"（跳过 {len(fetched)} 只已有数据）" if fetched else ""
print(f"正在抓取{freq}分钟K线 {sd} ~ {ed}，需抓取 {total} 只{skip_msg}...", 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)
bs_code = code_to_bs(code)
if not bs_code:
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)
with bs_query(
bs.query_history_k_data_plus,
bs_code,
"date,time,open,high,low,close,volume,amount",
start_date=sd, end_date=ed,
frequency=freq, adjustflag="3",
) as rs:
rows = []
while rs.next():
r = rs.get_row_data()
dt_str = _parse_datetime(r[0], r[1])
rows.append({
"code": code,
"datetime": dt_str,
"open": _clean(r[2]),
"high": _clean(r[3]),
"low": _clean(r[4]),
"close": _clean(r[5]),
"volume": _clean(r[6]),
"amount": _clean(r[7]),
})
if rows:
batch_upsert(model, rows, ["code", "datetime"])
success += 1
except Exception:
fail += 1
total_elapsed = time.time() - t_start
avg = total_elapsed / (i + 1)
eta = avg * (total - i - 1)
if (i + 1) % 100 == 0:
elapsed = time.time() - t_start
print(f" [{i+1}/{total}] 进度... 成功:{success} 失败:{fail} 已用时:{elapsed:.0f}s", flush=True)
print(f" [{i+1}/{total}] {code} 成功:{success} 失败:{fail} "
f"已用时:{total_elapsed:.0f}s 预计剩余:{eta:.0f}s", flush=True)
time.sleep(delay)
if not symbol:
time.sleep(delay)
total_time = time.time() - t_start
print(f" 分时行情抓取完成，成功:{success} 失败:{fail} 总耗时:{total_time:.1f}s", flush=True)
print(f" {freq}分钟K线抓取完成，成功:{success} 失败:{fail} 总耗时:{total_time:.1f}s", flush=True)
src/fetchers/market_daily.py
+103
View File
@@ -0,0 +1,103 @@
"""每日涨跌停统计 — 从 stock_daily 汇总
按板块区分：
- 10% 涨跌停：主板（沪60x/深00x）
- 20% 涨跌停：科创板（688/689）、创业板（300/301）
用法：
python -m src.main --market-daily
python -m src.main --market-daily --start-date 20260101 --end-date 20260509
"""
from datetime import datetime
from sqlalchemy import select, func, case, and_, text
from src.db import StockDaily, MarketDaily, batch_upsert, get_session
# 20% 板块：300/301 创业板，688/689 科创板
def _is_20pct(code: str) -> bool:
return code[:3] in ("300", "301", "688", "689")
def _fetch_history(start_date: str | None, end_date: str | None):
if end_date is None:
end_date = datetime.now().strftime("%Y%m%d")
if start_date is None:
start_date = "19901219"
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]}"
print(f"正在从 stock_daily 汇总涨跌停统计 {sd} ~ {ed}...", flush=True)
# 跳过已有的日期
session = get_session()
try:
existing = session.execute(
select(MarketDaily.date)
.where(MarketDaily.date >= sd)
.where(MarketDaily.date <= ed)
)
existing_dates = {str(row[0]) for row in existing}
finally:
session.close()
# 一次性查出所有需处理的代码前缀信息，按日聚合
session = get_session()
try:
# 用 SQL 直接按日期分组，区分10%/20%板块
sql = text("""
SELECT
date,
SUM(CASE WHEN pct_change >= 9.8
AND code NOT LIKE '300%%'
AND code NOT LIKE '301%%'
AND code NOT LIKE '688%%'
AND code NOT LIKE '689%%'
THEN 1 ELSE 0 END) AS limit_up_10,
SUM(CASE WHEN pct_change >= 19.5
AND (code LIKE '300%%' OR code LIKE '301%%'
OR code LIKE '688%%' OR code LIKE '689%%')
THEN 1 ELSE 0 END) AS limit_up_20,
SUM(CASE WHEN pct_change <= -9.8
AND code NOT LIKE '300%%'
AND code NOT LIKE '301%%'
AND code NOT LIKE '688%%'
AND code NOT LIKE '689%%'
THEN 1 ELSE 0 END) AS limit_down_10,
SUM(CASE WHEN pct_change <= -19.5
AND (code LIKE '300%%' OR code LIKE '301%%'
OR code LIKE '688%%' OR code LIKE '689%%')
THEN 1 ELSE 0 END) AS limit_down_20
FROM stock_daily
WHERE date >= :sd AND date <= :ed
AND pct_change IS NOT NULL
GROUP BY date
ORDER BY date
""")
result = session.execute(sql, {"sd": sd, "ed": ed})
finally:
session.close()
rows = []
for row in result:
date_str = str(row.date)
if date_str in existing_dates:
continue
rows.append({
"date": date_str,
"limit_up_10": int(row.limit_up_10 or 0),
"limit_up_20": int(row.limit_up_20 or 0),
"limit_down_10": int(row.limit_down_10 or 0),
"limit_down_20": int(row.limit_down_20 or 0),
})
if rows:
batch_upsert(MarketDaily, rows, ["date"])
print(f" 已写入 {len(rows)} 天涨跌停统计（{rows[0]['date']} ~ {rows[-1]['date']}）", flush=True)
else:
print(" 无新数据", flush=True)
def fetch_market_daily(start_date: str | None = None, end_date: str | None = None):
fetch_history(start_date, end_date)
src/fetchers/money_flow.py
-125
View File
@@ -1,125 +0,0 @@
"""个股资金流向抓取模块 — 使用东方财富数据源
通过 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
src/fetchers/sector.py
+152
View File
@@ -0,0 +1,152 @@
"""行业+地域分类数据抓取
数据源：
- 行业分类：BaoStock query_stock_industry()（证监会行业分类）
- 地域分类：东方财富 F10 CompanySurveyAjax（省份）
用法：
python -m src.main --sector
python -m src.main --sector --industry-only # 仅行业
python -m src.main --sector --region-only # 仅地域
"""
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
import baostock as bs
from src.baostock_conn import bs_query, bs_login
from src.db import StockInfo, StockSector, batch_upsert, get_session
from sqlalchemy import select
_HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
}
_F10_URL = "https://emweb.securities.eastmoney.com/PC_HSF10/CompanySurvey/CompanySurveyAjax"
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_industry() -> dict[str, str]:
"""从 BaoStock 获取全部股票的行业分类"""
result = {}
with bs_query(bs.query_stock_industry) as rs:
while rs.next():
r = rs.get_row_data()
bs_code = r[1]
industry = r[3]
code = bs_code.split(".")[1] if "." in bs_code else bs_code
if industry:
result[code] = industry
return result
def _fetch_one_region(code: str) -> tuple[str, str | None]:
"""获取单只股票的省份"""
prefix = "SH" if code.startswith(("6", "9")) else "SZ"
try:
r = requests.get(
_F10_URL,
params={"code": f"{prefix}{code}"},
headers=_HEADERS,
timeout=8,
)
jbzl = r.json().get("jbzl", {})
return code, jbzl.get("qy")
except Exception:
return code, None
def _fetch_region_batch(codes: list[str], workers: int = 10) -> dict[str, str]:
"""并发获取省份数据"""
result = {}
total = len(codes)
done = 0
t_start = time.time()
with ThreadPoolExecutor(max_workers=workers) as pool:
futures = {pool.submit(_fetch_one_region, c): c for c in codes}
for future in as_completed(futures):
code, region = future.result()
done += 1
if region:
result[code] = region
if done % 500 == 0:
elapsed = time.time() - t_start
print(f" [{done}/{total}] 地域数据... 已获取:{len(result)} 已用时:{elapsed:.0f}s", flush=True)
return result
def fetch_sector(industry_only: bool = False, region_only: bool = False):
"""抓取行业分类 + 地域分类"""
codes = _get_stock_codes()
if not codes:
print(" 无股票列表，请先运行 --stock-info", flush=True)
return
industry_map = {}
region_map = {}
# 1. 行业分类（BaoStock）
if not region_only:
bs_login()
print("正在抓取行业分类...", flush=True)
industry_map = _fetch_industry()
print(f" 行业分类: {len(industry_map)} 只股票有数据", flush=True)
# 2. 地域分类（东方财富 F10，仅获取缺失的）
if not industry_only:
session = get_session()
try:
existing = session.execute(
select(StockSector.code, StockSector.region)
.where(StockSector.region.isnot(None))
)
region_map = {row[0]: row[1] for row in existing}
finally:
session.close()
codes_need_region = [c for c in codes if c not in region_map]
if codes_need_region:
print(f"正在抓取地域分类，共 {len(codes_need_region)} 只（10并发）...", flush=True)
new_region = _fetch_region_batch(codes_need_region, workers=10)
region_map.update(new_region)
print(f" 地域分类: 共 {len(region_map)} 只股票有数据", flush=True)
else:
print(" 地域数据已完整，跳过", flush=True)
# 3. 加载已有数据，合并写入
existing_data = {}
session = get_session()
try:
rows_db = session.execute(select(StockSector)).scalars().all()
for r in rows_db:
existing_data[r.code] = {"industry": r.industry, "region": r.region}
finally:
session.close()
all_codes = set(existing_data.keys()) | set(industry_map.keys()) | set(region_map.keys())
if not all_codes:
print(" 无数据", flush=True)
return
rows = []
for code in all_codes:
existing = existing_data.get(code, {})
rows.append({
"code": code,
"industry": industry_map.get(code, existing.get("industry")),
"region": region_map.get(code, existing.get("region")),
})
batch_upsert(StockSector, rows, ["code"])
print(f" 已写入 {len(rows)} 条行业+地域记录", flush=True)
src/fetchers/stock_list.py
+26 -93
View File
@@ -1,102 +1,35 @@
"""股票列表抓取模块 — 分别从沪深北三个交易所获取股票基本信息
"""股票列表抓取模块 — 使用 BaoStock
AKShare 对三个交易所有独立的接口，返回字段名不一致：
- 沪市: 证券代码、证券简称、上市日期
- 深市: A股代码、A股简称、A股上市日期
- 北交所: 证券代码、证券简称、上市日期
需要分别处理字段映射，最后按 code 去重合并。
BaoStock query_stock_basic() 一次返回全部证券（含指数、基金等），
通过 type=1 过滤只保留股票，status=1 过滤仍在上市的。
注意：BaoStock 不含北交所（920xxx）。
"""
import time
import akshare as ak
from src.config import get_fetch_config
import baostock as bs
from src.baostock_conn import bs_query
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)
"""从 BaoStock 获取沪深A股列表，upsert 到 stock_info 表"""
print("正在抓取股票列表...", flush=True)
rows = []
with bs_query(bs.query_stock_basic) as rs:
while rs.next():
r = rs.get_row_data()
# fields: code, code_name, ipoDate, outDate, type, status
bs_code, name, ipo_date, out_date, typ, status = r[0], r[1], r[2], r[3], r[4], r[5]
if typ != "1" or status != "1":
continue
code = bs_code.split(".")[1] if "." in bs_code else bs_code
rows.append({
"code": code,
"name": name,
"ipo_date": ipo_date if ipo_date else None,
})
# 沪市
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]
if rows:
batch_upsert(StockInfo, rows, ["code"])
print(f" 股票列表已更新，共 {len(rows)} 只", flush=True)
else:
print(" 无股票数据", flush=True)
src/fetchers/trading_day.py
+55 -58
View File
@@ -1,102 +1,99 @@
"""交易日历模块 — 提供多级回退策略获取交易日列表
"""交易日历模块 — 使用 BaoStock
1. 本地 trading_day 表（最快，之前已缓存）
2. AKShare 新浪交易日历接口（数据准确，覆盖 1990-12-19 ~ 2026-12-31）
3. BaoStock 交易日历（备用，需过滤掉周末）
4. 从 stock_daily 表已有数据推断（最后手段）
交易日历是日线数据完整性判断的关键依据：
股票在某个日期范围内的记录数必须等于交易日数才算完整。
数据源优先级：
1. 本地 trading_day 表（最快，之前已缓存）
2. BaoStock 交易日历（需过滤掉周末）
3. 从 stock_daily 表已有数据推断（最后手段）
"""
import time
from datetime import datetime
import baostock as bs
import akshare as ak
from src.db import TradingDay, batch_upsert, get_session
from src.db import TradingDay, StockDaily, batch_upsert, get_session
from src.baostock_conn import bs_query
from sqlalchemy import select, func
def _fetch_akshare(sd: str, ed: str) -> list[str] | None:
"""从 AKShare 新浪接口获取交易日历（数据准确，覆盖1990~2026）"""
try:
df = ak.tool_trade_date_hist_sina()
df["date_str"] = df["trade_date"].astype(str)
mask = (df["date_str"] >= sd) & (df["date_str"] <= ed)
filtered = df.loc[mask]
days = filtered["date_str"].tolist()
return days if days else None
except Exception:
return None
def _fetch_baostock(sd: str, ed: str) -> list[str] | None:
"""从 BaoStock 获取交易日历，过滤掉周末（BaoStock 会返回包括周末在内的所有日期）"""
"""从 BaoStock 获取交易日历，过滤周末"""
try:
bs.login()
rs = bs.query_trade_dates(start_date=sd, end_date=ed)
days = []
while (rs.error_code == "0") and rs.next():
d = rs.get_row_data()[0]
# 过滤周末：weekday() 返回 0-6 (Mon-Sun)
if datetime.strptime(d, "%Y-%m-%d").weekday() < 5:
days.append(d)
bs.logout()
return days if days else None
with bs_query(bs.query_trade_dates, start_date=sd, end_date=ed) as rs:
days = []
while (rs.error_code == "0") and rs.next():
d = rs.get_row_data()[0]
if datetime.strptime(d, "%Y-%m-%d").weekday() < 5:
days.append(d)
return days if days else None
except Exception:
try:
bs.logout()
except Exception:
pass
return None
def _fetch_and_save(start_date: str, end_date: str) -> list[str]:
"""按优先级尝试数据源，获取交易日并保存到本地表"""
"""从 BaoStock 获取交易日并保存到本地表"""
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]}"
for name, fn in [("AKShare", _fetch_akshare), ("BaoStock", _fetch_baostock)]:
print(f" 正在从 {name} 获取交易日历 {sd} ~ {ed}...", flush=True)
days = fn(sd, ed)
if days:
rows = [{"date": d} for d in days]
batch_upsert(TradingDay, rows, ["date"])
print(f" 交易日历已保存，{len(days)} 个交易日", flush=True)
return days
print(f" 正在从 BaoStock 获取交易日历 {sd} ~ {ed}...", flush=True)
days = _fetch_baostock(sd, ed)
if days:
rows = [{"date": d} for d in days]
batch_upsert(TradingDay, rows, ["date"])
print(f" 交易日历已保存，{len(days)} 个交易日", flush=True)
return days
print(" 所有数据源均失败，将从已有行情数据推断", flush=True)
print(" BaoStock 获取失败，将从已有行情数据推断", flush=True)
return _infer_from_daily(start_date, end_date)
def _format_date(d: str) -> str:
"""YYYYMMDD → YYYY-MM-DD"""
return f"{d[:4]}-{d[4:6]}-{d[6:8]}"
def get_trading_days(start_date: str, end_date: str) -> list[str]:
"""获取指定范围内的交易日列表，本地表优先，缺失则从数据源拉取"""
"""获取指定范围内的交易日列表
优先查本地表，若本地数据未覆盖完整范围则从 BaoStock 补全。
"""
sd = _format_date(start_date)
ed = _format_date(end_date)
session = get_session()
try:
result = session.execute(
select(TradingDay.date)
.where(TradingDay.date >= start_date)
.where(TradingDay.date <= end_date)
.where(TradingDay.date >= sd)
.where(TradingDay.date <= ed)
.order_by(TradingDay.date)
)
days = [str(row[0]) for row in result]
if days:
if days and days[0] <= sd and days[-1] >= ed:
return days
finally:
session.close()
return _fetch_and_save(start_date, end_date)
# 本地数据不完整，从 BaoStock 获取
fetched = _fetch_and_save(start_date, end_date)
# 合并本地 + 新获取的数据去重
if not fetched:
return days if days else []
all_days = sorted(set(days + fetched))
return all_days
def _infer_from_daily(start_date: str, end_date: str) -> list[str]:
"""从 stock_daily 表推断交易日"""
from src.db import StockDaily
sd = _format_date(start_date)
ed = _format_date(end_date)
session = get_session()
try:
result = session.execute(
select(func.distinct(StockDaily.date))
.where(StockDaily.date >= start_date)
.where(StockDaily.date <= end_date)
.where(StockDaily.date >= sd)
.where(StockDaily.date <= ed)
.order_by(StockDaily.date)
)
return [str(row[0]) for row in result]
@@ -105,10 +102,10 @@ def _infer_from_daily(start_date: str, end_date: str) -> list[str]:
def fetch_trading_days(start_date: str | None = None, end_date: str | None = None):
"""独立抓取交易日历并保存到 trading_day 表
"""独立抓取交易日历并保存
用法：python -m src.main --trading-day --start-date 19901219 --end-date 20261231
默认从 1990-12-19（沪市开市日）到今天，建议一次性拉取完整日历。
默认从 1990-12-19（沪市开市日）到今天。
"""
if end_date is None:
end_date = time.strftime("%Y%m%d")
src/main.py
+52 -44
View File
@@ -1,80 +1,88 @@
"""A股数据抓取工具主入口
"""A股数据抓取工具主入口 — BaoStock 数据源
用法示例：
python -m src.main --stock-list # 先抓取股票列表
python -m src.main --trading-day --start-date 20200101 --end-date 20261231
python -m src.main --stock-info # 先抓取股票列表
python -m src.main --trading-day # 抓取交易日历
python -m src.main --daily --start-date 20260501 --end-date 20260508
python -m src.main --all # 全量抓取
python -m src.main --financial --symbol 000001 # 单只股票财务数据
python -m src.main --financial --symbol 000001
python -m src.main --dividend
python -m src.main --intraday # 默认5分钟，近30天
python -m src.main --intraday --freq all # 全部频率
python -m src.main --intraday --start-date 20260508 --end-date 20260509
python -m src.main --intraday --symbol 000001 --freq 30
python -m src.main --sector # 行业+地域分类
python -m src.main --sector --industry-only # 仅行业分类
"""
import argparse
from datetime import datetime
from src.config import load_config
from src.db import init_db
from src.baostock_conn import bs_login, bs_logout
def main():
parser = argparse.ArgumentParser(description="A股数据抓取工具")
parser.add_argument("--stock-list", action="store_true", help="抓取股票列表")
parser = argparse.ArgumentParser(description="A股数据抓取工具（BaoStock）")
parser.add_argument("--stock-info", action="store_true", help="抓取股票列表")
parser.add_argument("--trading-day", 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("--financial", 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("--intraday", action="store_true", help="抓取分钟K线行情")
parser.add_argument("--freq", type=str, default="5",
choices=["5", "15", "30", "60", "all"],
help="分钟K线频率（默认5，all=全部）")
parser.add_argument("--sector", action="store_true", help="抓取行业+地域分类")
parser.add_argument("--industry-only", action="store_true", help="仅抓取行业分类")
parser.add_argument("--region-only", 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.trading_day, args.daily, args.financial, args.money_flow,
args.dragon_tiger, args.dividend, args.intraday, args.all]):
if not any([args.stock_info, args.trading_day, args.daily,
args.financial, args.dividend, args.intraday,
args.sector, args.industry_only, args.region_only]):
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()
bs_login()
try:
if args.stock_info:
from src.fetchers.stock_list import fetch_stock_list
fetch_stock_list()
if args.all or args.trading_day:
from src.fetchers.trading_day import fetch_trading_days
fetch_trading_days(start_date=args.start_date, end_date=args.end_date)
if args.trading_day:
from src.fetchers.trading_day import fetch_trading_days
fetch_trading_days(start_date=args.start_date, end_date=args.end_date)
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.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.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.dividend:
from src.fetchers.dividend import fetch_dividend
fetch_dividend(symbol=args.symbol)
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.intraday:
from src.fetchers.intraday import fetch_intraday
fetch_intraday(start_date=args.start_date, end_date=args.end_date,
symbol=args.symbol, freq=args.freq)
if args.all or args.dividend:
from src.fetchers.dividend import fetch_dividend
fetch_dividend(symbol=args.symbol)
if args.sector or args.industry_only or args.region_only:
from src.fetchers.sector import fetch_sector
fetch_sector(industry_only=args.industry_only, region_only=args.region_only)
if args.all or args.intraday:
from src.fetchers.intraday import fetch_intraday
fetch_intraday()
print("全部任务完成", flush=True)
print("全部任务完成", flush=True)
finally:
bs_logout()
if __name__ == "__main__":