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 18:33:41 +08:00
parent 4572deb264
commit 284ccb9b02
4 changed files with 238 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/db.py
+13
View File
@@ -262,6 +262,19 @@ class StockSector(Base):
region = Column(String(20), comment="省份/地域")
class StockConcept(Base):
__tablename__ = "stock_concept"
__table_args__ = (
UniqueConstraint("code", "concept_code", name="uq_concept_code"),
Index("ix_concept_code", "concept_code"),
)
id = Column(Integer, primary_key=True, autoincrement=True)
code = Column(String(10), nullable=False, comment="股票代码")
concept_code = Column(String(20), nullable=False, comment="概念板块代码")
concept_name = Column(String(100), nullable=False, comment="概念板块名称")
# ── 数据库连接管理 ──
_engine = None
_SessionFactory = None
src/fetchers/daily.py
+4 -15
View File
@@ -9,6 +9,7 @@ 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, batch_upsert, get_session, get_stock_codes, get_ipo_dates
from src.fetchers.trading_day import get_trading_days
from sqlalchemy import select, func, text
@@ -179,25 +180,13 @@ def fetch_daily(start_date: str | None = None, end_date: str | None = None):
bs_login()
# 直接查本地 trading_day 表，不调 BaoStock
sd_fmt = f"{start_date[:4]}-{start_date[4:6]}-{start_date[6:8]}"
ed_fmt = f"{end_date[:4]}-{end_date[4:6]}-{end_date[6:8]}"
from src.db import TradingDay
session = get_session()
try:
td_result = session.execute(
select(TradingDay.date)
.where(TradingDay.date >= sd_fmt)
.where(TradingDay.date <= ed_fmt)
.order_by(TradingDay.date)
)
trading_days = [str(row[0]) for row in td_result]
finally:
session.close()
# 优先使用本地交易日历；若覆盖不完整则自动补齐
trading_days = get_trading_days(start_date, end_date)
td_count = len(trading_days)
print(f" 交易日历: {start_date} ~ {end_date} 共 {td_count} 个交易日", flush=True)
# 排除未上市股票
ed_fmt = f"{end_date[:4]}-{end_date[4:6]}-{end_date[6:8]}"
session = get_session()
try:
not_listed = {row[0] for row in session.execute(
src/fetchers/sector.py
+213 -65
View File
@@ -1,15 +1,227 @@
"""行业+地域分类数据抓取
"""行业+地域+概念板块数据抓取
数据源：
- 行业分类：BaoStock query_stock_industry()（证监会行业分类）
- 地域分类：东方财富 F10 CompanySurveyAjax（省份）
- 概念板块：东方财富 stock_board_concept_name_em() 全量概念列表 + 每个概念成分股
用法：
python -m src.main --sector
python -m src.main --sector --industry-only # 仅行业
python -m src.main --sector --region-only # 仅地域
python -m src.main --sector --concept-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 StockSector, StockConcept, batch_upsert, get_session, get_stock_codes
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"
_EM_CONCEPT_LIST_URL = "https://push2.eastmoney.com/api/qt/clist/get"
_EM_CONCEPT_STOCKS_URL = "https://push2.eastmoney.com/api/qt/clist/get"
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_concept_list() -> list[dict]:
"""从东方财富获取全部概念板块列表"""
params = {
"pn": 1, "pz": 3000, "po": 1, "np": 1,
"ut": "bd1d9ddb04089700cf9c27f6f7426281",
"fltt": 2, "invt": 2,
"fid": "f3",
"fs": "m:90+t:3",
"fields": "f2,f3,f12,f14",
"_": int(time.time() * 1000),
}
try:
r = requests.get(_EM_CONCEPT_LIST_URL, params=params, headers=_HEADERS, timeout=15)
data = r.json().get("data", {}) or {}
items = data.get("diff", []) or []
return [{"code": item["f12"], "name": item["f14"]} for item in items if item.get("f12") and item.get("f14")]
except Exception as e:
print(f" 获取概念板块列表失败: {e}", flush=True)
return []
def _fetch_concept_stocks(concept_code: str) -> list[str]:
"""获取单个概念板块的成分股代码列表"""
params = {
"pn": 1, "pz": 5000, "po": 1, "np": 1,
"ut": "bd1d9ddb04089700cf9c27f6f7426281",
"fltt": 2, "invt": 2,
"fid": "f3",
"fs": f"b:{concept_code}+f:!50",
"fields": "f12",
"_": int(time.time() * 1000),
}
try:
r = requests.get(_EM_CONCEPT_STOCKS_URL, params=params, headers=_HEADERS, timeout=15)
data = r.json().get("data", {}) or {}
items = data.get("diff", []) or []
return [item["f12"] for item in items if item.get("f12")]
except Exception:
return []
def fetch_concept():
"""抓取全部概念板块及成分股，写入 stock_concept 表"""
print("正在获取概念板块列表...", flush=True)
concepts = _fetch_concept_list()
if not concepts:
print(" 概念板块列表获取失败", flush=True)
return
print(f" 共 {len(concepts)} 个概念板块，开始抓取成分股...", flush=True)
t_start = time.time()
rows = []
for i, concept in enumerate(concepts):
stocks = _fetch_concept_stocks(concept["code"])
for code in stocks:
rows.append({
"code": code,
"concept_code": concept["code"],
"concept_name": concept["name"],
})
if (i + 1) % 50 == 0:
elapsed = time.time() - t_start
print(f" [{i+1}/{len(concepts)}] 已处理 耗时:{elapsed:.0f}s", flush=True)
time.sleep(0.05)
if rows:
batch_upsert(StockConcept, rows, ["code", "concept_code"])
print(f" 概念板块写入完成，{len(concepts)} 个概念，{len(rows)} 条记录，耗时:{time.time()-t_start:.0f}s", flush=True)
else:
print(" 无概念板块数据", flush=True)
def fetch_sector(industry_only: bool = False, region_only: bool = False, concept_only: bool = False):
"""抓取行业分类 + 地域分类 + 概念板块"""
codes = get_stock_codes()
if not codes:
print(" 无股票列表，请先运行 --stock-info", flush=True)
return
if concept_only:
fetch_concept()
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)
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
@@ -77,67 +289,3 @@ def _fetch_region_batch(codes: list[str], workers: int = 10) -> dict[str, str]:
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/main.py
+8 -4
View File
@@ -12,13 +12,15 @@
python -m src.main --intraday --symbol 000001 --freq 30
python -m src.main --sector # 行业+地域分类
python -m src.main --sector --industry-only # 仅行业分类
python -m src.main --sector --concept-only # 仅概念板块
python -m src.main --index # 指数日线（上证/沪深300/创业板等）
"""
import argparse
from src.baostock_conn import bs_login, bs_logout
from src.config import load_config
from src.db import init_db
from src.baostock_conn import bs_login, bs_logout
def main():
@@ -36,6 +38,7 @@ def main():
parser.add_argument("--index", 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("--concept-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="指定单只股票代码")
@@ -45,7 +48,7 @@ def main():
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,
args.index]):
args.concept_only, args.index]):
parser.print_help()
return
@@ -79,9 +82,10 @@ def main():
fetch_intraday(start_date=args.start_date, end_date=args.end_date,
symbol=args.symbol, freq=args.freq)
if args.sector or args.industry_only or args.region_only:
if args.sector or args.industry_only or args.region_only or args.concept_only:
from src.fetchers.sector import fetch_sector
fetch_sector(industry_only=args.industry_only, region_only=args.region_only)
fetch_sector(industry_only=args.industry_only, region_only=args.region_only,
concept_only=args.concept_only)
if args.index:
from src.fetchers.index import fetch_index