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-12 00:29:52 +08:00
parent 947ed6dc78
commit 4572deb264
4 changed files with 118 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+24 -10
View File
@@ -1,6 +1,6 @@
# ashare-data
A股数据抓取工具，使用 [AKShare](https://github.com/akfamily/akshare) 获取数据，保存到 MySQL 数据库。
A股数据抓取工具，使用 [BaoStock](http://baostock.com) 获取数据，保存到 MySQL 数据库。
## 功能概览
@@ -9,6 +9,8 @@ A股数据抓取工具，使用 [AKShare](https://github.com/akfamily/akshare)
| 股票列表 | 沪深A股代码、名称、上市日期 | BaoStock |
| 交易日历 | 1990年至今的交易日列表 | BaoStock |
| 日线行情 | 开高低收、成交量/额、振幅、涨跌幅、换手率（前复权） | BaoStock |
| 指数日线 | 上证/深证/创业板等主要指数日线 | BaoStock |
| 涨跌停统计 | 每日主板/科创板/创业板涨跌停数量 | stock_daily 汇总 |
| 季频财务指标 | 盈利能力、偿债能力、现金流（最近8个季度） | BaoStock |
| 分红送转 | 每10股送转、派息、除权除息日（最近10年） | BaoStock |
| 分时行情 | 5/15/30/60分钟K线（开高低收、成交量/额） | BaoStock |
@@ -21,7 +23,7 @@ A股数据抓取工具，使用 [AKShare](https://github.com/akfamily/akshare)
### 1. 环境要求
- Python >= 3.10
- MySQL >= 5.7（建议 8.0+）
- MySQL >= 8.0
### 2. 安装依赖
@@ -66,16 +68,26 @@ CREATE DATABASE ashare CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
# 1. 先抓取股票列表（其他模块依赖此数据）
python -m src.main --stock-info
# 2. 抓取交易日历
# 2. 抓取交易日历（覆盖全历史，只需执行一次；此后每年末执行一次延长至未来）
python -m src.main --trading-day --start-date 19901219 --end-date 20261231
# 3. 抓取最近30天的日线行情
python -m src.main --daily
# 指定日期范围抓取日线
# 3. 抓取全历史日线行情（首次，耗时较长）
python -m src.main --daily --start-date 19901201 --end-date 20260511
python -m src.main --daily --start-date 19920101 --end-date 20260511
# 4. 日常增量更新日线（默认最近30天）
python -m src.main --daily
# 抓取指数日线
python -m src.main --index
# 抓取指数日线（指定日期范围）
python -m src.main --index --start-date 19901219 --end-date 20260511
# 汇总涨跌停统计（依赖 stock_daily 数据）
python -m src.main --market-daily
# 汇总涨跌停统计（指定日期范围）
python -m src.main --market-daily --start-date 20260101 --end-date 20260511
# 抓取财务指标（全部股票，最近8个季度）
python -m src.main --financial
@@ -105,7 +117,9 @@ python -m src.main --sector
选项:
--stock-info 抓取A股股票列表
--trading-day 抓取交易日历
--daily 抓取日线行情
--daily 抓取日线行情（增量，已完整自动跳过）
--index 抓取主要指数日线
--market-daily 汇总每日涨跌停统计（从 stock_daily 聚合）
--financial 抓取季频财务指标
--dividend 抓取分红送转数据
--intraday 抓取分钟K线行情
@@ -114,7 +128,7 @@ python -m src.main --sector
--industry-only 仅抓取行业分类
--region-only 仅抓取地域分类
日期过滤（对日线行情、交易日历、分钟K线生效）:
日期过滤（对日线行情、交易日历、分钟K线、指数、涨跌停统计生效）:
--start-date 开始日期，格式 YYYYMMDD
--end-date 结束日期，格式 YYYYMMDD
src/baostock_conn.py
+35 -3
View File
@@ -11,6 +11,8 @@ import baostock as bs
_lock = threading.Lock()
_logged_in = False
QUERY_TIMEOUT = 60 # 单次查询超时秒数
def bs_login():
"""全局只 login 一次"""
@@ -30,9 +32,21 @@ def bs_logout():
_logged_in = False
def _relogin():
"""超时后重连"""
global _logged_in
try:
bs.logout()
except Exception:
pass
_logged_in = False
bs.login()
_logged_in = True
@contextmanager
def bs_query(query_fn, *args, **kwargs):
"""加锁执行 BaoStock 查询，yield ResultData
"""加锁执行 BaoStock 查询，yield ResultData，超时自动重连
用法:
with bs_query(bs.query_history_k_data_plus, code, fields, ...) as rs:
@@ -45,9 +59,27 @@ def bs_query(query_fn, *args, **kwargs):
sig_parts += [f"{k}={v!r}" for k, v in kwargs.items()]
sig = ", ".join(sig_parts)
print(f" [{datetime.now().strftime('%H:%M:%S')}] [BS] {short_name}({sig})", flush=True)
result_box = [None]
exc_box = [None]
def _run():
try:
result_box[0] = query_fn(*args, **kwargs)
except Exception as e:
exc_box[0] = e
with _lock:
rs = query_fn(*args, **kwargs)
yield rs
t = threading.Thread(target=_run, daemon=True)
t.start()
t.join(timeout=QUERY_TIMEOUT)
if t.is_alive():
print(f" [{datetime.now().strftime('%H:%M:%S')}] [BS] 查询超时({QUERY_TIMEOUT}s)，重连...", flush=True)
_relogin()
raise TimeoutError(f"BaoStock query timeout: {short_name}({sig})")
if exc_box[0] is not None:
raise exc_box[0]
yield result_box[0]
def code_to_bs(code: str) -> str:
src/db.py
+14
View File
@@ -266,6 +266,7 @@ class StockSector(Base):
_engine = None
_SessionFactory = None
_stock_codes_cache: list[str] | None = None
_ipo_dates_cache: dict[str, str] | None = None
def get_stock_codes() -> list[str]:
@@ -281,6 +282,19 @@ def get_stock_codes() -> list[str]:
return _stock_codes_cache
def get_ipo_dates() -> dict[str, str]:
"""获取全部股票上市日期（进程内缓存，避免重复查询）"""
global _ipo_dates_cache
if _ipo_dates_cache is None:
session = get_session()
try:
result = session.execute(select(StockInfo.code, StockInfo.ipo_date))
_ipo_dates_cache = {code: str(ipo) for code, ipo in result if ipo}
finally:
session.close()
return _ipo_dates_cache
def get_engine():
global _engine
if _engine is None:
src/fetchers/daily.py
+45 -71
View File
@@ -8,8 +8,8 @@ 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, StockDaily, StockNoData, batch_upsert, get_session, get_stock_codes
from sqlalchemy import select, func
from src.db import StockInfo, StockDaily, batch_upsert, get_session, get_stock_codes, get_ipo_dates
from sqlalchemy import select, func, text
def _clean(val):
@@ -22,70 +22,67 @@ def _clean(val):
def _analyze_gaps(codes: list[str], start_date: str, end_date: str,
trading_days: list[str]) -> dict[str, list[str]]:
"""按月分段分析缺口。用 COUNT 对比交易日数，不逐条加载。"""
"""一条SQL统计每只股票每月行情数，与交易日对比找缺口。"""
if not trading_days:
return {}
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]}"
# 上市日期
session = get_session()
try:
ipo_result = session.execute(
select(StockInfo.code, StockInfo.ipo_date)
.where(StockInfo.code.in_(codes))
)
ipo_dates: dict[str, str] = {}
for code, ipo in ipo_result:
if ipo:
ipo_dates[code] = str(ipo)
finally:
session.close()
# 按月分段
td_by_month: dict[str, list[str]] = {}
for d in trading_days:
key = d[:7] # "2026-05"
td_by_month.setdefault(key, []).append(d)
gap_codes: set[str] = set()
total_months = len(td_by_month)
t0 = time.time()
for idx, (month_key, month_days) in enumerate(sorted(td_by_month.items())):
m_start = month_days[0]
m_end = month_days[-1]
# 上市日期（进程内缓存，只查一次）
ipo_dates = get_ipo_dates()
print(f" [1/3] 上市日期查询完成 {len(ipo_dates)} 只 {time.time()-t0:.1f}s", flush=True)
session = get_session()
try:
cnt_result = session.execute(
select(StockDaily.code, func.count(StockDaily.id))
.where(StockDaily.code.in_(codes))
.where(StockDaily.date >= m_start)
.where(StockDaily.date <= m_end)
.group_by(StockDaily.code)
)
code_cnt = {row[0]: row[1] for row in cnt_result}
finally:
session.close()
# 按月统计每只股票行情数（一条SQL）
t1 = time.time()
session = get_session()
try:
sql = text("""
SELECT code, DATE_FORMAT(date, '%Y-%m') AS month, COUNT(*) AS cnt
FROM stock_daily
WHERE date >= :sd AND date <= :ed
GROUP BY code, DATE_FORMAT(date, '%Y-%m')
""")
result = session.execute(sql, {"sd": sd, "ed": ed})
# code_month_cnt[code][month] = count
code_month_cnt: dict[str, dict[str, int]] = {}
for code, month, cnt in result:
code_month_cnt.setdefault(code, {})[month] = cnt
finally:
session.close()
print(f" [2/3] 行情按月统计完成 {time.time()-t1:.1f}s", flush=True)
new_gaps = 0
# 按月对比找缺口
t2 = time.time()
td_by_month: dict[str, list[str]] = {}
for d in trading_days:
td_by_month.setdefault(d[:7], []).append(d)
gap_codes: set[str] = set()
no_ipo_codes: set[str] = set()
for month_key, month_days in sorted(td_by_month.items()):
for code in codes:
if code in gap_codes:
continue
ipo = ipo_dates.get(code)
if ipo and ipo > m_end:
if not ipo:
# stock_info 里没有该股票，跳过避免误判全量缺口
no_ipo_codes.add(code)
continue
expected = [d for d in month_days if not ipo or d >= ipo]
if ipo > month_days[-1]:
continue
expected = [d for d in month_days if d >= ipo]
if not expected:
continue
if code_cnt.get(code, 0) < len(expected):
cnt = code_month_cnt.get(code, {}).get(month_key, 0)
if cnt < len(expected):
gap_codes.add(code)
new_gaps += 1
# 调试：打印触发缺口的月份和计数
print(f" [gap] {code} {month_key} 本地:{cnt} 期望:{len(expected)}", flush=True)
print(f" [{idx+1}/{total_months}] {month_key} 交易日:{len(month_days)} 新增缺口:{new_gaps} "
f"累计:{len(gap_codes)} 已用时:{time.time()-t0:.1f}s", flush=True)
print(f" [3/3] 缺口对比完成 缺口股票:{len(gap_codes)} 只 跳过(无上市日期):{len(no_ipo_codes)} 只 {time.time()-t2:.1f}s", flush=True)
if not gap_codes:
return {}
@@ -116,25 +113,13 @@ def _analyze_gaps(codes: list[str], start_date: str, end_date: str,
gaps[code] = [front[0], front[-1]]
elif back:
gaps[code] = [back[0], back[-1]]
else:
gaps[code] = [expected[0], expected[-1]]
# else: 数据两端已覆盖，内部缺口属停牌，不重抓
else:
gaps[code] = [expected[0], expected[-1]]
return gaps
def _mark_suspensions(nodata_map: dict[str, list[str]]):
if not nodata_map:
return
rows = []
for code, dates in nodata_map.items():
rows.extend({"code": code, "date": d} for d in dates)
if rows:
batch_upsert(StockNoData, rows, ["code", "date"])
print(f" 标记停牌: {len(rows)} 条（{len(nodata_map)} 只股票）", flush=True)
def _fetch_baostock(code: str, start_date: str, end_date: str) -> list[dict] | None:
bs_code = code_to_bs(code)
if not bs_code:
@@ -243,7 +228,6 @@ def fetch_daily(start_date: str | None = None, end_date: str | None = None):
success = 0
fail = 0
nodata_count = 0
nodata_map: dict[str, list[str]] = {}
t_start = time.time()
for i, code in enumerate(gaps):
@@ -263,15 +247,7 @@ def fetch_daily(start_date: str | None = None, end_date: str | None = None):
print(f" {code} 写入失败: {e}", flush=True)
fail += 1
continue
returned_dates = {row["date"] for row in rows}
missing = [d for d in trading_days if d not in returned_dates
and g[0] <= d <= g[-1]]
if missing:
nodata_map[code] = missing
else:
missing = [d for d in trading_days if g[0] <= d <= g[-1]]
if missing:
nodata_map[code] = missing
nodata_count += 1
t_write = time.time() - t1
@@ -283,7 +259,5 @@ def fetch_daily(start_date: str | None = None, end_date: str | None = None):
f"成功:{success} 剩余:{eta:.0f}s", flush=True)
time.sleep(delay)
_mark_suspensions(nodata_map)
total_time = time.time() - t_start
print(f" 日线行情抓取完成，成功:{success} 失败:{fail} 停牌:{nodata_count} 总耗时:{total_time:.1f}s", flush=True)