diff --git a/backend/alembic/versions/20260905_01_etf_basic.py b/backend/alembic/versions/20260905_01_etf_basic.py new file mode 100644 index 0000000..b6565df --- /dev/null +++ b/backend/alembic/versions/20260905_01_etf_basic.py @@ -0,0 +1,41 @@ +"""etf_basic：场内 ETF 列表（东财快照 + 规模字段） + +Revision ID: 20260905_01 +Revises: 20260902_01 +Create Date: 2026-09-05 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "20260905_01" +down_revision: Union[str, Sequence[str], None] = "20260902_01" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "etf_basic", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("ts_code", sa.String(length=12), nullable=False), + sa.Column("symbol", sa.String(length=10), nullable=False), + sa.Column("name", sa.String(length=64), nullable=False), + sa.Column("exchange", sa.String(length=8), nullable=False), + sa.Column("list_date", sa.String(length=8), nullable=True), + sa.Column("total_mv", sa.Float(), nullable=True), + sa.Column("circ_mv", sa.Float(), nullable=True), + sa.Column("turnover_rate", sa.Float(), nullable=True), + sa.Column("updated_at", sa.DateTime(), nullable=True), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("ts_code", name="uq_etf_basic_ts_code"), + ) + op.create_index("ix_etf_basic_ts_code", "etf_basic", ["ts_code"]) + op.create_index("ix_etf_basic_symbol", "etf_basic", ["symbol"]) + + +def downgrade() -> None: + op.drop_index("ix_etf_basic_symbol", table_name="etf_basic") + op.drop_index("ix_etf_basic_ts_code", table_name="etf_basic") + op.drop_table("etf_basic") diff --git a/backend/alembic/versions/20260906_01_stock_company.py b/backend/alembic/versions/20260906_01_stock_company.py new file mode 100644 index 0000000..66faf01 --- /dev/null +++ b/backend/alembic/versions/20260906_01_stock_company.py @@ -0,0 +1,48 @@ +"""stock_company：上市公司基本信息（tushare 按需懒加载 + 30 天新鲜度 + 墓碑负缓存） + +Revision ID: 20260906_01 +Revises: 20260905_01 +Create Date: 2026-09-06 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "20260906_01" +down_revision: Union[str, Sequence[str], None] = "20260905_01" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "stock_company", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("ts_code", sa.String(length=12), nullable=False), + sa.Column("com_name", sa.String(length=255), nullable=True), + sa.Column("com_id", sa.String(length=32), nullable=True), + sa.Column("chairman", sa.String(length=128), nullable=True), + sa.Column("manager", sa.String(length=128), nullable=True), + sa.Column("secretary", sa.String(length=128), nullable=True), + sa.Column("reg_capital", sa.Float(), nullable=True), + sa.Column("setup_date", sa.String(length=8), nullable=True), + sa.Column("province", sa.String(length=32), nullable=True), + sa.Column("city", sa.String(length=32), nullable=True), + sa.Column("introduction", sa.Text(), nullable=True), + sa.Column("website", sa.String(length=255), nullable=True), + sa.Column("email", sa.String(length=255), nullable=True), + sa.Column("office", sa.String(length=255), nullable=True), + sa.Column("employees", sa.Integer(), nullable=True), + sa.Column("main_business", sa.Text(), nullable=True), + sa.Column("business_scope", sa.Text(), nullable=True), + sa.Column("updated_at", sa.DateTime(), nullable=True), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("ts_code", name="uq_stock_company_ts_code"), + ) + op.create_index("ix_stock_company_ts_code", "stock_company", ["ts_code"]) + + +def downgrade() -> None: + op.drop_index("ix_stock_company_ts_code", table_name="stock_company") + op.drop_table("stock_company") diff --git a/backend/alembic/versions/20260907_01_stock_finance_dividend.py b/backend/alembic/versions/20260907_01_stock_finance_dividend.py new file mode 100644 index 0000000..c9dd3b5 --- /dev/null +++ b/backend/alembic/versions/20260907_01_stock_finance_dividend.py @@ -0,0 +1,92 @@ +"""stock_financial / stock_dividend / stock_sync_state：个股财务与分红懒加载管道 + +Revision ID: 20260907_01 +Revises: 20260906_01 +Create Date: 2026-09-07 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "20260907_01" +down_revision: Union[str, Sequence[str], None] = "20260906_01" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "stock_financial", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("ts_code", sa.String(length=12), nullable=False), + sa.Column("end_date", sa.String(length=8), nullable=False), + sa.Column("ann_date", sa.String(length=8), nullable=True), + sa.Column("eps", sa.Float(), nullable=True), + sa.Column("bps", sa.Float(), nullable=True), + sa.Column("ocfps", sa.Float(), nullable=True), + sa.Column("roe", sa.Float(), nullable=True), + sa.Column("roe_dt", sa.Float(), nullable=True), + sa.Column("grossprofit_margin", sa.Float(), nullable=True), + sa.Column("netprofit_margin", sa.Float(), nullable=True), + sa.Column("debt_to_assets", sa.Float(), nullable=True), + sa.Column("or_yoy", sa.Float(), nullable=True), + sa.Column("netprofit_yoy", sa.Float(), nullable=True), + sa.Column("dt_netprofit_yoy", sa.Float(), nullable=True), + sa.Column("profit_dedt", sa.Float(), nullable=True), + sa.Column("rd_exp", sa.Float(), nullable=True), + sa.Column("total_revenue", sa.Float(), nullable=True), + sa.Column("operate_profit", sa.Float(), nullable=True), + sa.Column("n_income_attr_p", sa.Float(), nullable=True), + sa.Column("total_assets", sa.Float(), nullable=True), + sa.Column("total_hldr_eqy", sa.Float(), nullable=True), + sa.Column("n_cashflow_act", sa.Float(), nullable=True), + sa.Column("updated_at", sa.DateTime(), nullable=True), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("ts_code", "end_date", name="uq_stock_financial_code_end"), + ) + op.create_index("ix_stock_financial_ts_code", "stock_financial", ["ts_code"]) + + op.create_table( + "stock_dividend", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("ts_code", sa.String(length=12), nullable=False), + sa.Column("end_date", sa.String(length=8), nullable=True), + sa.Column("ann_date", sa.String(length=8), nullable=True), + sa.Column("div_proc", sa.String(length=16), nullable=True), + sa.Column("stk_div", sa.Float(), nullable=True), + sa.Column("stk_bo_rate", sa.Float(), nullable=True), + sa.Column("stk_co_rate", sa.Float(), nullable=True), + sa.Column("cash_div", sa.Float(), nullable=True), + sa.Column("cash_div_tax", sa.Float(), nullable=True), + sa.Column("base_share", sa.Float(), nullable=True), + sa.Column("record_date", sa.String(length=8), nullable=True), + sa.Column("ex_date", sa.String(length=8), nullable=True), + sa.Column("pay_date", sa.String(length=8), nullable=True), + sa.Column("div_listdate", sa.String(length=8), nullable=True), + sa.Column("imp_ann_date", sa.String(length=8), nullable=True), + sa.Column("updated_at", sa.DateTime(), nullable=True), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index("ix_stock_dividend_ts_code", "stock_dividend", ["ts_code"]) + + op.create_table( + "stock_sync_state", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("ts_code", sa.String(length=12), nullable=False), + sa.Column("kind", sa.String(length=16), nullable=False), + sa.Column("last_synced_at", sa.DateTime(), nullable=True), + sa.Column("has_data", sa.Boolean(), nullable=True), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("ts_code", "kind", name="uq_stock_sync_code_kind"), + ) + op.create_index("ix_stock_sync_state_ts_code", "stock_sync_state", ["ts_code"]) + + +def downgrade() -> None: + op.drop_index("ix_stock_sync_state_ts_code", table_name="stock_sync_state") + op.drop_table("stock_sync_state") + op.drop_index("ix_stock_dividend_ts_code", table_name="stock_dividend") + op.drop_table("stock_dividend") + op.drop_index("ix_stock_financial_ts_code", table_name="stock_financial") + op.drop_table("stock_financial") diff --git a/backend/alembic/versions/20260907_02_stock_reference.py b/backend/alembic/versions/20260907_02_stock_reference.py new file mode 100644 index 0000000..2cebdea --- /dev/null +++ b/backend/alembic/versions/20260907_02_stock_reference.py @@ -0,0 +1,34 @@ +"""stock_reference：个股参考数据（tushare 参考数据版块，按股按分类存 JSON 快照懒加载） + +Revision ID: 20260907_02 +Revises: 20260907_01 +Create Date: 2026-09-07 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "20260907_02" +down_revision: Union[str, Sequence[str], None] = "20260907_01" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "stock_reference", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("ts_code", sa.String(length=12), nullable=False), + sa.Column("kind", sa.String(length=24), nullable=False), + sa.Column("rows_json", sa.Text(), nullable=True), + sa.Column("updated_at", sa.DateTime(), nullable=True), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("ts_code", "kind", name="uq_stock_reference_code_kind"), + ) + op.create_index("ix_stock_reference_ts_code", "stock_reference", ["ts_code"]) + + +def downgrade() -> None: + op.drop_index("ix_stock_reference_ts_code", table_name="stock_reference") + op.drop_table("stock_reference") diff --git a/backend/app/data/company.py b/backend/app/data/company.py new file mode 100644 index 0000000..e9436fa --- /dev/null +++ b/backend/app/data/company.py @@ -0,0 +1,178 @@ +"""上市公司基本信息（tushare stock_company，详情页按需单查懒加载）。 + +不做批量同步：详情页打开才触发，ts_code= 单查一次一调用； +行即缓存 —— stock_company 表 30 天新鲜度门控（数据月更），tushare 查无此股 +写墓碑行（业务字段全 NULL）做负缓存，避免无数据代码每次都穿透（0.35s 控频 ++ 可能 62s 限频重试）。墓碑同样按 updated_at 参与 30 天刷新，新股上市后能自动补上。 +""" +from __future__ import annotations + +import asyncio +import time +from datetime import datetime, timedelta, timezone + +from sqlalchemy import select +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from ..config import settings +from ..db import async_session +from ..models import StockCompany + +_REFRESH_DAYS = 30 + +# 频率超限特征（等待 62s 重试一次；与 screener.market_sync / data.etf_sync._call_retry 同款语义） +_RATE_MARKS = ("频率超限", "每分钟") + +# 显式列出全部字段：introduction/office/main_business/business_scope 文档标注默认不显示， +# 不传 fields 时 tushare 不返回这四列（实测 000001.SZ） +_FIELDS = ( + "ts_code,com_name,com_id,chairman,manager,secretary,reg_capital," + "setup_date,province,city,introduction,website,email,office," + "employees,main_business,business_scope" +) + +_pro = None # 惰性单例（get_pro 每次都 ts.set_token 写文件，没必要重复） + + +def _get_pro(): + if not settings.tushare_token: + raise RuntimeError("未配置 TUSHARE_TOKEN，无法拉取公司简介（backend/.env）") + global _pro + if _pro is None: + from .tushare_provider import get_pro + + _pro = get_pro() + return _pro + + +def _call_retry(fn, *args, **kwargs): + """同步调用 tushare 接口；「每分钟」级频率超限等 62s 重试一次。""" + try: + return fn(*args, **kwargs) + except Exception as e: # noqa: BLE001 + msg = str(e) + if any(m in msg for m in _RATE_MARKS) and "小时" not in msg: + time.sleep(62) + return fn(*args, **kwargs) + raise + + +def _utcnow() -> datetime: + return datetime.now(timezone.utc).replace(tzinfo=None) + + +def _fresh(updated_at: datetime | None) -> bool: + return updated_at is not None and updated_at >= _utcnow() - timedelta(days=_REFRESH_DAYS) + + +def _s(v) -> str | None: + """pandas NaN / 空串 / None -> None，其余 strip。""" + if v is None or (isinstance(v, float) and v != v): + return None + s = str(v).strip() + return s or None + + +def _f(v) -> float | None: + try: + f = float(v) + except (TypeError, ValueError): + return None + return None if f != f else f # NaN -> None + + +def _i(v) -> int | None: + f = _f(v) + return None if f is None else int(f) + + +def fetch_company_sync(ts_code: str) -> dict | None: + """同步拉单只公司简介（需在 to_thread 里跑）；返回行 dict，无此股返回 None。""" + time.sleep(settings.screener_sync_interval) + df = _call_retry(_get_pro().stock_company, ts_code=ts_code, fields=_FIELDS) + if df is None or df.empty: + return None + r = df.iloc[0] + return { + "ts_code": ts_code, + "com_name": _s(r.get("com_name")), + "com_id": _s(r.get("com_id")), + "chairman": _s(r.get("chairman")), + "manager": _s(r.get("manager")), + "secretary": _s(r.get("secretary")), + "reg_capital": _f(r.get("reg_capital")), + "setup_date": _s(r.get("setup_date")), + "province": _s(r.get("province")), + "city": _s(r.get("city")), + "introduction": _s(r.get("introduction")), + "website": _s(r.get("website")), + "email": _s(r.get("email")), + "office": _s(r.get("office")), + "employees": _i(r.get("employees")), + "main_business": _s(r.get("main_business")), + "business_scope": _s(r.get("business_scope")), + "updated_at": _utcnow(), + } + + +_COLS = ( + "ts_code", "com_name", "com_id", "chairman", "manager", "secretary", + "reg_capital", "setup_date", "province", "city", "introduction", + "website", "email", "office", "employees", "main_business", "business_scope", +) + + +def _row_dict(row: StockCompany) -> dict: + return {c: getattr(row, c) for c in _COLS} + + +async def _upsert(session: AsyncSession, row: dict) -> None: + stmt = pg_insert(StockCompany).values(row) + stmt = stmt.on_conflict_do_update( + index_elements=["ts_code"], + set_={c: stmt.excluded[c] for c in row if c != "ts_code"}, + ) + await session.execute(stmt) + await session.commit() + + +# per-code 锁：同 code 首次并发 N 个请求只有 1 个打 tushare，其余等锁后双检命中； +# dict 不清理（对象数 = 全市场股票数，内存可忽略）。uvicorn 单进程场景够用， +# 多 worker 最坏情况是重复拉一次 + ON CONFLICT 幂等，无害。 +_code_locks: dict[str, asyncio.Lock] = {} +_guard = asyncio.Lock() + + +async def get_company(session: AsyncSession, ts_code: str) -> dict | None: + """读穿透：新鲜行直返；否则加 per-code 锁 -> 新会话双检 -> to_thread 拉 -> upsert。 + + 返回 None = 确认无数据（墓碑已落库）；抛异常 = tushare 拉取失败且无旧行可降级。 + """ + row = (await session.execute( + select(StockCompany).where(StockCompany.ts_code == ts_code))).scalar_one_or_none() + if row is not None and _fresh(row.updated_at): + return _row_dict(row) if row.com_name is not None else None # 墓碑 -> None + # 释放请求会话持有的连接：后面可能隔着 1-2s 的 tushare 调用，别长占连接池。 + # 用 close() 而非 rollback()：rollback 会把会话身份映射里的实例全部 expire—— + # 包括 require_user 刚塞进 auth._session_cache 的 User，下个请求命中鉴权缓存即 + # DetachedInstanceError 500；close() 同样归还连接且已加载属性保持可访问。 + await session.close() + + async with _guard: + lock = _code_locks.setdefault(ts_code, asyncio.Lock()) + async with lock: + async with async_session() as s2: # 锁内重读 + 写入走新会话 + row = (await s2.execute( + select(StockCompany).where(StockCompany.ts_code == ts_code))).scalar_one_or_none() + if row is not None and _fresh(row.updated_at): + return _row_dict(row) if row.com_name is not None else None + try: + fetched = await asyncio.to_thread(fetch_company_sync, ts_code) + except Exception: + # 降级：库内有真实旧行（哪怕超 30 天）照常返回，不把「上游挂了」伪装成「无数据」 + if row is not None and row.com_name is not None: + return _row_dict(row) + raise + await _upsert(s2, fetched or {"ts_code": ts_code, "updated_at": _utcnow()}) + return fetched diff --git a/backend/app/data/dividend.py b/backend/app/data/dividend.py new file mode 100644 index 0000000..b4b783e --- /dev/null +++ b/backend/app/data/dividend.py @@ -0,0 +1,110 @@ +"""个股分红送股（tushare dividend 全历史，详情页按需单查懒加载）。 + +每 code 同步时全量替换（dividend 无稳定唯一业务键，每票行数几十条，替换最简单）； +stock_sync_state(kind='dividend') 7 天新鲜度门控，空结果也是有效墓碑 +（相当多股票从不分红，避免每次点击都穿透控频调用）。 +""" +from __future__ import annotations + +import asyncio +import time + +from sqlalchemy import delete, select +from sqlalchemy.ext.asyncio import AsyncSession + +from ..config import settings +from ..db import async_session +from ..models import StockDividend +from .sync_utils import call_retry, f_clean, fresh, get_pro_lazy, read_sync_state, s_clean, upsert_sync_state, utcnow + +_REFRESH_DAYS = 7 + +_FIELDS = ("ts_code,end_date,ann_date,div_proc,stk_div,stk_bo_rate,stk_co_rate," + "cash_div,cash_div_tax,base_share,record_date,ex_date,pay_date," + "div_listdate,imp_ann_date") + +_COLS = ("ts_code", "end_date", "ann_date", "div_proc", "stk_div", "stk_bo_rate", + "stk_co_rate", "cash_div", "cash_div_tax", "base_share", "record_date", + "ex_date", "pay_date", "div_listdate", "imp_ann_date") + + +def fetch_dividends_sync(ts_code: str) -> list[dict]: + """同步拉全历史分红（需在 to_thread 里跑）；无分红返回空列表。""" + time.sleep(settings.screener_sync_interval) + df = call_retry(get_pro_lazy().dividend, ts_code=ts_code, fields=_FIELDS) + if df is None or df.empty: + return [] + # tushare 会返回完全重复的行（实测 000001.SZ 同一除权日出现两次）； + # 前端按 ex_date 聚合求和，不去重会双计分红金额 + df = df.drop_duplicates() + rows: list[dict] = [] + for _, r in df.iterrows(): + rows.append({ + "ts_code": ts_code, + "end_date": s_clean(r.get("end_date")), + "ann_date": s_clean(r.get("ann_date")), + "div_proc": s_clean(r.get("div_proc")), + "stk_div": f_clean(r.get("stk_div")), + "stk_bo_rate": f_clean(r.get("stk_bo_rate")), + "stk_co_rate": f_clean(r.get("stk_co_rate")), + "cash_div": f_clean(r.get("cash_div")), + "cash_div_tax": f_clean(r.get("cash_div_tax")), + "base_share": f_clean(r.get("base_share")), + "record_date": s_clean(r.get("record_date")), + "ex_date": s_clean(r.get("ex_date")), + "pay_date": s_clean(r.get("pay_date")), + "div_listdate": s_clean(r.get("div_listdate")), + "imp_ann_date": s_clean(r.get("imp_ann_date")), + "updated_at": utcnow(), + }) + rows.sort(key=lambda x: ((x.get("end_date") or "", x.get("ann_date") or "")), reverse=True) + return rows + + +async def _read_rows(session: AsyncSession, ts_code: str) -> list[dict]: + rs = (await session.execute( + select(StockDividend).where(StockDividend.ts_code == ts_code) + .order_by(StockDividend.end_date.desc(), StockDividend.ann_date.desc()))).scalars().all() + return [{c: getattr(r, c) for c in _COLS} for r in rs] + + +async def _replace(session: AsyncSession, ts_code: str, rows: list[dict]) -> None: + """全量替换 + 状态落库（一个事务，由本函数 commit）。""" + await session.execute(delete(StockDividend).where(StockDividend.ts_code == ts_code)) + session.add_all([StockDividend(**r) for r in rows]) + await upsert_sync_state(session, ts_code, "dividend", has_data=bool(rows)) + await session.commit() + + +# per-code 锁：同 code 首次并发 N 个请求只有 1 个打 tushare，其余等锁后双检命中 +_code_locks: dict[str, asyncio.Lock] = {} +_guard = asyncio.Lock() + + +async def get_dividends(session: AsyncSession, ts_code: str) -> list[dict]: + """读穿透：同步状态新鲜直返库内行；否则 per-code 锁 -> 新会话双检 -> to_thread 拉 -> 全量替换。 + + 返回空列表 = 确认无分红（墓碑已落库）；抛异常 = tushare 拉取失败且无旧行可降级。 + """ + state = await read_sync_state(session, ts_code, "dividend") + if state is not None and fresh(state.last_synced_at, _REFRESH_DAYS): + return await _read_rows(session, ts_code) + # 释放请求会话连接（同 finance.get_finance：close() 而非 rollback()，防鉴权缓存 User 被 expire） + await session.close() + + async with _guard: + lock = _code_locks.setdefault(ts_code, asyncio.Lock()) + async with lock: + async with async_session() as s2: + state = await read_sync_state(s2, ts_code, "dividend") + if state is not None and fresh(state.last_synced_at, _REFRESH_DAYS): + return await _read_rows(s2, ts_code) + old = await _read_rows(s2, ts_code) + try: + fetched = await asyncio.to_thread(fetch_dividends_sync, ts_code) + except Exception: + if old: # 降级：返旧行，不把「上游挂了」伪装成「无分红」 + return old + raise + await _replace(s2, ts_code, fetched) + return fetched diff --git a/backend/app/data/etf_provider.py b/backend/app/data/etf_provider.py new file mode 100644 index 0000000..35c1368 --- /dev/null +++ b/backend/app/data/etf_provider.py @@ -0,0 +1,109 @@ +"""东方财富 ETF 快照（免 token 直连 HTTP，仅用于 etf_basic 列表与规模字段）。 + +K 线数据不走这里：统一走 Tushare fund_daily（见 etf_sync / fetcher）， +东财在链路里只承担一件 Tushare quicksync 镜像做不到的事—— +全市场 ETF 名单 + 总市值/流通市值/换手率（镜像上 fund_etf_basic 不存在）。 + +接口注意：clist 实测 pz 上限 100（传 50000 也只回 100），必须按 pn 翻页 +拿全 ~1600 只；push2 主站短连发几次会直接断连，push2delay 镜像稳， +列表/市值用延迟值无妨（价格另有 K 线）。 +""" +from __future__ import annotations + +import asyncio + +import httpx + +from .symbols import is_etf_symbol + +_HEADERS = { + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36", +} + +# 东财 ETF 板块（akshare fund_etf_spot_em 同款集合：股票/混合/债券/货币/商品/跨境等） +_SPOT_HOSTS = [ + "https://push2delay.eastmoney.com", + "https://push2.eastmoney.com", +] +_SPOT_PARAMS = { + "pn": "1", + "pz": "100", + "po": "1", + "np": "1", + "ut": "bd1d9ddb04089700cf9c27f6f7426281", + "fltt": "2", + "invt": "2", + "fid": "f12", + "fs": "b:MK0021,b:MK0022,b:MK0023,b:MK0024,b:MK0026,b:MK0027,b:MK0028,b:MK0029", + # f12 代码 / f13 市场(1沪0深) / f14 名称 / f8 换手% / f20 总市值 / f21 流通市值（元） + "fields": "f12,f13,f14,f8,f20,f21", +} + + +def _num(v) -> float | None: + """东财 fltt=2 下停牌/缺数据的字段是 '-' 字符串。""" + if v is None or isinstance(v, str): + return None + f = float(v) + return None if f != f else f + + +def new_client() -> httpx.AsyncClient: + """统一构造（超时/UA）；同步工厂，`async with new_client() as c` 使用。""" + return httpx.AsyncClient(timeout=httpx.Timeout(15.0), headers=_HEADERS) + + +async def _fetch_spot_page(client: httpx.AsyncClient, host: str, pn: int): + params = {**_SPOT_PARAMS, "pn": str(pn)} + resp = await client.get(f"{host}/api/qt/clist/get", params=params) + resp.raise_for_status() + data = resp.json().get("data") or {} + return data.get("total") or 0, data.get("diff") or [] + + +async def fetch_etf_spot(client: httpx.AsyncClient) -> list[dict]: + """全市场场内 ETF 快照（翻页拿全 ~1600 只）-> [{ts_code, symbol, name, exchange, + turnover_rate, total_mv, circ_mv}]。空结果视为异常（接口改版/被拦截时宁可不覆盖表）。""" + diff: list[dict] = [] + for host in _SPOT_HOSTS: + try: + total, first = await _fetch_spot_page(client, host, 1) + diff = first + pn = 2 + while total and len(diff) < total: + await asyncio.sleep(0.15) # 翻页间隔，礼貌控频 + _, page = await _fetch_spot_page(client, host, pn) + if not page: + break + diff.extend(page) + pn += 1 + break # 首个可用 host 拿完即止 + except Exception: # noqa: BLE001 —— 主镜像抖动换备用镜像整重来 + diff = [] + continue + rows: list[dict] = [] + for d in diff: + code = str(d.get("f12") or "").strip() + name = str(d.get("f14") or "").strip() + market = d.get("f13") + if not code or not name or market is None: + continue + if not is_etf_symbol(code): + continue # 板块返回里混进的 LOF/封基（16/50/57 开头）不进 ETF 表 + exchange = "SH" if int(market) == 1 else "SZ" + rows.append({ + "ts_code": f"{code}.{exchange}", + "symbol": code, + "name": name, + "exchange": exchange, + "turnover_rate": _num(d.get("f8")), + "total_mv": _num(d.get("f20")), + "circ_mv": _num(d.get("f21")), + }) + if not rows: + raise RuntimeError("东财 ETF 快照为空（接口可能改版或被限流）") + return rows + + +__all__ = ["fetch_etf_spot", "new_client"] diff --git a/backend/app/data/etf_sync.py b/backend/app/data/etf_sync.py new file mode 100644 index 0000000..1cd4327 --- /dev/null +++ b/backend/app/data/etf_sync.py @@ -0,0 +1,369 @@ +"""ETF 全市场数据同步（列表走东财快照，K 线走 Tushare fund_daily）。 + +分工： +- 东财 push2 clist：全市场 ETF 名单 + 总市值/流通市值/换手率（quicksync 镜像上 + fund_etf_basic 不存在，规模字段无替代源）-> etf_basic； +- Tushare fund_daily（quicksync 可用，与股票 daily 同源同控频）-> candles + 不复权底座，单位换算与股票一致（vol 手->份 ×100、amount 千元->元 ×1000）。 + +同步策略（全串行，Tushare 按分钟限频，并发无意义）： +- 逐日模式：交易日历里尚无任何 ETF 日线的日期，一天一调用拿全市场基金日线 + （过滤到 etf_basic 符号），日常增量通常只有当天 1 次调用； +- 逐只模式：无任何缓存的 ETF（新上市/历史缺口）按 ts_code 全量拉取，每次 1 调用； + full=true 时对所有 ETF 重拉（修数/回补 amount 用）。 + +- 进程内后台任务（与 screener.market_sync 同款模式），前端轮询 /api/etf/sync/status； +- 复权：quicksync 无 fund_adj_factor，ETF 暂无因子，qfq/hfq 切换时按无因子 + 原样返回（api._adjust_bars 的既有语义）。 +""" +from __future__ import annotations + +import asyncio +import time +from datetime import datetime, timedelta, timezone + +from sqlalchemy import delete, func, select, text +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from .. import cache +from ..config import settings +from . import etf_provider + +# 进程内单例任务状态（uvicorn 单进程场景够用） +_state: dict = { + "running": False, + "step": None, + "total": 0, # 本次需处理的单元数（缺失交易日 + 需拉取的 ETF 只数） + "done": 0, + "error": None, + "started_at": None, + "finished_at": None, +} +_task: asyncio.Task | None = None +_lock = asyncio.Lock() + +_BATCH = 3000 # upsert 分批行数（asyncpg 单语句参数上限 32766，10 列/行） +# fund_daily 返回全市场基金 ~2100 行，一天一批远小于上限 + +# 频率超限特征（等待 62s 重试一次；与 screener.market_sync._call_retry 同款语义） +_RATE_MARKS = ("频率超限", "每分钟") + + +def _call_retry(fn, *args, **kwargs): + """同步调用 tushare 接口；「每分钟」级频率超限等 62s 重试一次。""" + try: + return fn(*args, **kwargs) + except Exception as e: # noqa: BLE001 + msg = str(e) + if any(m in msg for m in _RATE_MARKS) and "小时" not in msg: + time.sleep(62) + return fn(*args, **kwargs) + raise + + +def _get_pro(): + """token 检查 + 返回 pro api 客户端（同步对象，调用需 to_thread 包裹）。""" + if not settings.tushare_token: + raise RuntimeError("未配置 TUSHARE_TOKEN，无法同步 ETF 日线（backend/.env）") + from .tushare_provider import get_pro + + return get_pro() + + +def _utcnow() -> datetime: + return datetime.now(timezone.utc).replace(tzinfo=None) + + +def _parse_d(s: str) -> datetime: + return datetime.strptime(str(s), "%Y%m%d") + + +async def _sync_spot(session: AsyncSession) -> int: + """快照 -> etf_basic（upsert + 删除已退市），返回列表行数。""" + from ..models import EtfBasic + + async with etf_provider.new_client() as client: + rows = await etf_provider.fetch_etf_spot(client) + now = _utcnow() + stmt = pg_insert(EtfBasic).values([{**r, "updated_at": now} for r in rows]) + stmt = stmt.on_conflict_do_update( + index_elements=["ts_code"], + set_={ + "name": stmt.excluded.name, + "exchange": stmt.excluded.exchange, + "total_mv": stmt.excluded.total_mv, + "circ_mv": stmt.excluded.circ_mv, + "turnover_rate": stmt.excluded.turnover_rate, + "updated_at": stmt.excluded.updated_at, + }, + ) + await session.execute(stmt) + # 快照外的 ETF 已退市（东财列表不再返回） + await session.execute(delete(EtfBasic).where(EtfBasic.ts_code.not_in({r["ts_code"] for r in rows}))) + await session.commit() + return len(rows) + + +def _fetch_day_sync(pro, d: str) -> list[dict]: + """拉某交易日全市场场内基金日线（fund_daily；未生成的日期返回空）。""" + time.sleep(settings.screener_sync_interval) + df = _call_retry(pro.fund_daily, trade_date=d) + if df is None or df.empty: + return [] + rows = [] + for _, r in df.iterrows(): + amt = r.get("amount") + rows.append({ + "ts": _parse_d(d), + "ts_code": r["ts_code"], + "open": float(r["open"]), "high": float(r["high"]), + "low": float(r["low"]), "close": float(r["close"]), + "vol": float(r["vol"]), # 手 + "amount": (float(amt) if amt is not None and amt == amt else None), # 千元 + }) + return rows + + +def _fetch_symbol_sync(pro, ts_code: str, start: str | None, end: str | None) -> list[dict]: + """按 ts_code 增量/全量拉单只 ETF 日线（start=None 即上市以来全量）。""" + time.sleep(settings.screener_sync_interval) + df = _call_retry(pro.fund_daily, ts_code=ts_code, start_date=start, end_date=end) + if df is None or df.empty: + return [] + df = df.sort_values("trade_date") + rows = [] + for _, r in df.iterrows(): + amt = r.get("amount") + rows.append({ + "ts": _parse_d(r["trade_date"]), + "ts_code": ts_code, + "open": float(r["open"]), "high": float(r["high"]), + "low": float(r["low"]), "close": float(r["close"]), + "vol": float(r["vol"]), + "amount": (float(amt) if amt is not None and amt == amt else None), + }) + return rows + + +def _upsert_candles_stmt(batch: list[dict]): + from ..models import Candle + + stmt = pg_insert(Candle).values(batch) + return stmt.on_conflict_do_update( + index_elements=["symbol", "timeframe", "ts"], + set_={"open": stmt.excluded.open, "high": stmt.excluded.high, + "low": stmt.excluded.low, "close": stmt.excluded.close, + "volume": stmt.excluded.volume, + # 增量源缺额时保留库内旧值（与股票底座同语义） + "amount": func.coalesce(stmt.excluded.amount, Candle.amount)}, + ) + + +def _day_batch(rows: list[dict], sym_set: set[str], d_str: str) -> list[dict]: + """某日 fund_daily 行 -> candles 批（过滤到 etf_basic 符号；手->份、千元->元）。""" + return [ + {"symbol": r["ts_code"].split(".")[0], "timeframe": "1d", + "ts": _parse_d(d_str), + "open": r["open"], "high": r["high"], "low": r["low"], "close": r["close"], + "volume": r["vol"] * 100.0, + "amount": (r["amount"] * 1000.0) if r["amount"] is not None else None, + "turnover": None} + for r in rows if r["ts_code"].split(".")[0] in sym_set + ] + + +def _symbol_batch(symbol: str, rows: list[dict]) -> list[dict]: + return [ + {"symbol": symbol, "timeframe": "1d", "ts": r["ts"], + "open": r["open"], "high": r["high"], "low": r["low"], "close": r["close"], + "volume": r["vol"] * 100.0, + "amount": (r["amount"] * 1000.0) if r["amount"] is not None else None, + "turnover": None} + for r in rows + ] + + +async def _write_batches(session: AsyncSession, batch: list[dict]) -> None: + for i in range(0, len(batch), _BATCH): + await session.execute(_upsert_candles_stmt(batch[i : i + _BATCH])) + await session.commit() + + +async def _backfill_list_dates(session: AsyncSession, dates: dict[str, str]) -> None: + """首根 K 线日回填 list_date（仅空缺处）：一条 UPDATE ... FROM (VALUES)。""" + pairs = list(dates.items()) + for i in range(0, len(pairs), 1000): + chunk = pairs[i : i + 1000] + vals = ", ".join(f"(:s{j}, :d{j})" for j in range(len(chunk))) + params = {f"s{j}": s for j, (s, _) in enumerate(chunk)} + params.update({f"d{j}": d for j, (_, d) in enumerate(chunk)}) + await session.execute(text( + f"UPDATE etf_basic e SET list_date = v.d FROM (VALUES {vals}) AS v(symbol, d) " + "WHERE e.symbol = v.symbol AND e.list_date IS NULL" + ), params) + + +async def _run_sync(full: bool) -> None: + from ..db import async_session + from ..models import Candle, EtfBasic, TradeCalendar + + try: + pro = await asyncio.to_thread(_get_pro) + + # 1) 快照 -> etf_basic + _state["step"] = "正在拉取 ETF 列表" + async with async_session() as session: + n_list = await _sync_spot(session) + etfs = (await session.execute( + select(EtfBasic.ts_code, EtfBasic.symbol, EtfBasic.exchange, EtfBasic.list_date) + )).all() + # 零缓存 ETF（新上市/历史缺口，需逐只全量拉）： + # NOT EXISTS 走 (symbol,timeframe,ts) 索引探测，1514 次 ms 级； + # 比对「全 ETF 符号 GROUP BY max(ts)」（扫数百万索引行）便宜得多 + have_any = set((await session.execute(text( + "SELECT e.symbol FROM etf_basic e WHERE EXISTS (SELECT 1 FROM candles c " + "WHERE c.symbol = e.symbol AND c.timeframe = '1d')" + ))).scalars()) + # 已落库的 ETF 交易日（YYYYMMDD）：逐日模式的跳过依据。 + # 只看近 40 天（索引范围扫）——更早的历史缺口由 full 全量重拉兜底， + # 全表 distinct 对千万行 candles 表要几十秒，不能每次同步都付 + fresh = not have_any or full + have_dates: set[str] = set() if fresh else { + r.strftime("%Y%m%d") for r in (await session.execute( + select(func.distinct(func.date(Candle.ts))).where( + Candle.timeframe == "1d", + Candle.symbol.in_(select(EtfBasic.symbol)), + Candle.ts >= datetime.now() - timedelta(days=40), + ) + )).scalars() if r is not None + } + cal = (await session.execute( + select(TradeCalendar.trade_date).order_by(TradeCalendar.trade_date.desc()) + )).scalars().all() + + sym_set = {e.symbol for e in etfs} + today = datetime.now().strftime("%Y%m%d") + since40 = (datetime.now() - timedelta(days=40)).strftime("%Y%m%d") + + # 2) 任务编排：逐日模式补近窗缺失交易日（fresh 库改走逐只全量）； + # 逐只模式拉零缓存 ETF，full=true 全量重拉 + todo_dates = [] if fresh else [ + d for d in cal if since40 <= d <= today and d not in have_dates + ] + per_symbol = list(etfs) if fresh else [ + e for e in etfs if e.symbol not in have_any + ] + _state["total"] = len(todo_dates) + len(per_symbol) + _state["done"] = 0 + fails: list[str] = [] + new_list_dates: dict[str, str] = {} + written_rows = 0 # 实际写入的 K 线行数（决定是否作废 candles 相关缓存） + + # 3) 逐日模式：一天一调用（交易日历空时跳过——由逐只模式兜底） + for d in sorted(todo_dates): + _state["step"] = f"正在同步 {d} 日线（{_state['done'] + 1}/{_state['total']}）" + try: + rows = await asyncio.to_thread(_fetch_day_sync, pro, d) + if rows: + batch = _day_batch(rows, sym_set, d) + if batch: + written_rows += len(batch) + async with async_session() as session: + await _write_batches(session, batch) + except Exception as ex: # noqa: BLE001 —— 单日失败不拖垮整体 + fails.append(f"{d}: {str(ex)[:80]}") + _state["done"] += 1 + + # 4) 逐只模式：零缓存 ETF 全量拉取（start=None 即上市以来；full 同理） + for ts_code, symbol, _exch, has_list_date in per_symbol: + _state["step"] = f"正在同步 ETF 日线 {symbol}（{_state['done'] + 1}/{_state['total']}）" + try: + rows = await asyncio.to_thread(_fetch_symbol_sync, pro, ts_code, None, None) + except Exception as ex: # noqa: BLE001 + fails.append(f"{ts_code}: {str(ex)[:80]}") + _state["done"] += 1 + continue + if rows: + # 全量拉取的首根 = 真实上市日 + if not has_list_date: + new_list_dates[symbol] = rows[0]["ts"].strftime("%Y%m%d") + batch = _symbol_batch(symbol, rows) + written_rows += len(batch) + try: + async with async_session() as session: + await _write_batches(session, batch) + except Exception as ex: # noqa: BLE001 + fails.append(f"{ts_code}: {str(ex)[:80]}") + _state["done"] += 1 + + # 5) list_date 回填（一次 SQL） + if new_list_dates: + async with async_session() as session: + await _backfill_list_dates(session, new_list_dates) + await session.commit() + + # candles 已更新：作废旧 K 线预览缓存（只在真的写了行时——空跑不作废， + # 免得每次同步都触发一轮 >10s 的统计重聚合）；etf 版本号作废 ETF 列表缓存 + if written_rows: + await cache.bump_version("candles") + await cache.bump_version("etf") + + parts = [f"同步完成（{n_list} 只 ETF"] + if todo_dates: + parts.append(f"{len(todo_dates)} 个交易日") + if per_symbol: + parts.append(f"{len(per_symbol)} 只逐只补数") + _state["step"] = "，".join(parts) + ")" + (f"，{len(fails)} 项失败" if fails else "") + if fails: + _state["error"] = "部分失败：" + "；".join(fails[:3]) + ("…" if len(fails) > 3 else "") + except Exception as e: # noqa: BLE001 + _state["error"] = f"同步失败：{str(e)[:300]}" + _state["step"] = "同步失败" + finally: + _state["running"] = False + _state["finished_at"] = datetime.now() + + +async def start_sync(full: bool = False) -> dict: + """幂等启动后台同步；已在跑则直接返回当前状态。full=true 所有 ETF 全量重拉。""" + global _task + async with _lock: + if _state["running"] and _task and not _task.done(): + return dict(_state) + _state.update({ + "running": True, "step": "准备同步", "total": 0, "done": 0, + "error": None, "started_at": datetime.now(), "finished_at": None, + }) + _task = asyncio.create_task(_run_sync(full)) + return dict(_state) + + +async def get_status(session: AsyncSession) -> dict: + """任务状态 + DB 实况。行情侧只做单符号 max(ts) 索引探测（candles 是千万行表， + 全表聚合 >10s，绝不能落在轮询热路径上），取最早上市且已有日线的一只当「数据更新至」。""" + from ..models import Candle, EtfBasic + + etfs = int(await session.scalar(select(func.count()).select_from(EtfBasic)) or 0) + # 探测样本：优先最早上市（历史最长最稳）且已有日线的一只（EXISTS 走索引，ms 级） + probe = await session.scalar( + text(""" + SELECT e.symbol FROM etf_basic e + WHERE EXISTS (SELECT 1 FROM candles c + WHERE c.symbol = e.symbol AND c.timeframe = '1d') + ORDER BY e.list_date NULLS LAST, e.symbol LIMIT 1 + """) + ) + last = None + if probe: + last = await session.scalar( + select(func.max(Candle.ts)).where(Candle.symbol == probe, Candle.timeframe == "1d") + ) + status = dict(_state) + status.update({ + "stats": {"etfs": etfs}, + "last_trade_date": last, + "last_synced_at": _state.get("finished_at") or _state.get("started_at"), + "ready": etfs > 0 and last is not None, + }) + return status diff --git a/backend/app/data/finance.py b/backend/app/data/finance.py new file mode 100644 index 0000000..6e29418 --- /dev/null +++ b/backend/app/data/finance.py @@ -0,0 +1,167 @@ +"""个股财务数据（fina_indicator 财务指标 + 三大报表关键值，详情页按需单查懒加载）。 + +四源按报告期合并进 stock_financial 宽表（一行 = 一个报告期）： + fina_indicator（doc 79，无 report_type 概念） + + income / balancesheet / cashflow（report_type=1 合并报表） +近五年窗口（start_date = 当年-5年 的 0101）；季更数据 7 天新鲜度门控。 +每源独立容错：部分接口失败不拖垮整体，缺失列靠 upsert 列级 coalesce 保旧值， +7 天后下次过期刷新自动重试失败源。四源全失败且无旧行 -> 抛异常（上游故障）。 +stock_sync_state(kind='finance') 做新鲜度与墓碑（无数据股票不重复穿透控频调用）。 +""" +from __future__ import annotations + +import asyncio +import time +from datetime import date + +from sqlalchemy import func, select +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from ..config import settings +from ..db import async_session +from ..models import StockFinancial +from .sync_utils import call_retry, f_clean, fresh, get_pro_lazy, read_sync_state, s_clean, upsert_sync_state, utcnow + +_REFRESH_DAYS = 7 + +# 四源显式字段（财务接口列多，必须显式传 fields；金额单位元、比率百分数沿用 tushare 原始） +_FI_FIELDS = ("ts_code,ann_date,end_date,eps,bps,ocfps,roe,roe_dt,grossprofit_margin," + "netprofit_margin,debt_to_assets,or_yoy,netprofit_yoy,dt_netprofit_yoy," + "profit_dedt,rd_exp") +_INC_FIELDS = "ts_code,ann_date,end_date,total_revenue,operate_profit,n_income_attr_p" +_BS_FIELDS = "ts_code,ann_date,end_date,total_assets,total_hldr_eqy_exc_min_int" +_CF_FIELDS = "ts_code,ann_date,end_date,n_cashflow_act" + +# tushare 字段 -> 模型列名（仅 balancesheet 归母权益超长名改短） +_COL_MAP = {"total_hldr_eqy_exc_min_int": "total_hldr_eqy"} + +_SOURCES = ( + ("fina_indicator", _FI_FIELDS, {}), + ("income", _INC_FIELDS, {"report_type": "1"}), + ("balancesheet", _BS_FIELDS, {"report_type": "1"}), + ("cashflow", _CF_FIELDS, {"report_type": "1"}), +) + +_COLS = ( + "ts_code", "end_date", "ann_date", + "eps", "bps", "ocfps", "roe", "roe_dt", "grossprofit_margin", "netprofit_margin", + "debt_to_assets", "or_yoy", "netprofit_yoy", "dt_netprofit_yoy", "profit_dedt", "rd_exp", + "total_revenue", "operate_profit", "n_income_attr_p", + "total_assets", "total_hldr_eqy", "n_cashflow_act", +) +_VALUE_COLS = _COLS[3:] # ann_date 之外的其余业务列（批量 upsert 需统一键集） + + +def fetch_finance_sync(ts_code: str) -> dict[str, dict]: + """拉近五年四源财务数据并按报告期合并（同步网络 IO，需在 to_thread 里跑）。 + + 返回 {end_date: 行dict}；四源全部失败且无任何数据时抛异常。 + """ + pro = get_pro_lazy() + start = f"{date.today().year - 5}0101" # 含整五个年度的年报 + merged: dict[str, dict] = {} + failed = 0 + for api, fields, kwargs in _SOURCES: + time.sleep(settings.screener_sync_interval) + try: + df = call_retry(getattr(pro, api), ts_code=ts_code, start_date=start, fields=fields, **kwargs) + except Exception: # noqa: BLE001 单源失败不拖垮整体（缺列保旧值，7 天后自动重试） + failed += 1 + continue + if df is None or df.empty: + continue + # 同一报告期 tushare 可能返回多条（调整前/后副本），按公告日升序取最后一条 + df = df.sort_values("ann_date", na_position="first").drop_duplicates("end_date", keep="last") + for _, r in df.iterrows(): + end = s_clean(r.get("end_date")) + if not end: + continue + row = merged.setdefault(end, {"ts_code": ts_code, "end_date": end, "ann_date": None}) + ann = s_clean(r.get("ann_date")) + if ann: + row["ann_date"] = ann + for col in fields.split(","): + if col in ("ts_code", "ann_date", "end_date"): + continue + v = f_clean(r.get(col)) + if v is not None: + row[_COL_MAP.get(col, col)] = v + if failed == len(_SOURCES) and not merged: + raise RuntimeError(f"tushare 财务四源均失败: {ts_code}") + return merged + + +def _row_dict(row: StockFinancial) -> dict: + return {c: getattr(row, c) for c in _COLS} + + +async def _read_rows(session: AsyncSession, ts_code: str) -> list[dict]: + rs = (await session.execute( + select(StockFinancial).where(StockFinancial.ts_code == ts_code) + .order_by(StockFinancial.end_date.desc()))).scalars().all() + return [_row_dict(r) for r in rs] + + +async def _upsert_rows(session: AsyncSession, rows: list[dict]) -> None: + """批量 upsert（不 commit）；列级 coalesce 保旧值——失败源的列、tushare 改为空值的列都保留库内旧值。""" + now = utcnow() + full = [] + for r in rows: + row = {c: None for c in _VALUE_COLS} # 统一键集：批量 insert 要求各 dict 同构 + row.update(r) + row["updated_at"] = now + full.append(row) + stmt = pg_insert(StockFinancial).values(full) + set_ = {c: func.coalesce(stmt.excluded[c], getattr(StockFinancial, c)) for c in ("ann_date", *_VALUE_COLS)} + set_["updated_at"] = stmt.excluded.updated_at + stmt = stmt.on_conflict_do_update(index_elements=["ts_code", "end_date"], set_=set_) + await session.execute(stmt) + + +# per-code 锁：同 code 首次并发 N 个请求只有 1 个打 tushare，其余等锁后双检命中 +# （uvicorn 单进程场景够用；多 worker 最坏情况是重复拉一次 + ON CONFLICT 幂等，无害） +_code_locks: dict[str, asyncio.Lock] = {} +_guard = asyncio.Lock() + + +async def get_finance(session: AsyncSession, ts_code: str) -> list[dict] | None: + """读穿透：同步状态新鲜直返；否则 per-code 锁 -> 新会话双检 -> to_thread 拉四源 -> upsert。 + + 返回 None = 确认无财务数据（墓碑已落库，多为新股/退市老股）； + 抛异常 = 四源均失败且无旧行可降级。 + """ + state = await read_sync_state(session, ts_code, "finance") + if state is not None and fresh(state.last_synced_at, _REFRESH_DAYS): + if not state.has_data: + return None + return await _read_rows(session, ts_code) + # 释放请求会话持有的连接：后面可能隔着数秒的 tushare 调用，别长占连接池。 + # 用 close() 而非 rollback()：rollback 会把会话身份映射里的实例全部 expire—— + # 包括 require_user 刚塞进 auth._session_cache 的 User（expire_on_commit=False + # 只保 commit，不保 rollback），下个请求命中鉴权缓存即 DetachedInstanceError 500。 + # close() 同样归还连接，且已加载属性保持可访问（脱管安全，与鉴权缓存约定一致）。 + await session.close() + + async with _guard: + lock = _code_locks.setdefault(ts_code, asyncio.Lock()) + async with lock: + async with async_session() as s2: # 锁内重读 + 写入走新会话 + state = await read_sync_state(s2, ts_code, "finance") + if state is not None and fresh(state.last_synced_at, _REFRESH_DAYS): + if not state.has_data: + return None + return await _read_rows(s2, ts_code) + old = await _read_rows(s2, ts_code) + try: + merged = await asyncio.to_thread(fetch_finance_sync, ts_code) + except Exception: + # 降级：库内有旧行（哪怕超 7 天）照常返回，不把「上游挂了」伪装成「无数据」 + if old: + return old + raise + if merged: + await _upsert_rows(s2, list(merged.values())) + await upsert_sync_state(s2, ts_code, "finance", has_data=bool(merged) or bool(old)) + await s2.commit() + return await _read_rows(s2, ts_code) diff --git a/backend/app/data/index_global.py b/backend/app/data/index_global.py new file mode 100644 index 0000000..90126a6 --- /dev/null +++ b/backend/app/data/index_global.py @@ -0,0 +1,420 @@ +"""指数专题数据源（tushare 指数接口族）。 + +- 列表层：21 个国际指数最新收盘 + 45 日 spark（pro.index_global），SWR 整包缓存 + （模式同 market_overview：进程内新鲜期直返 -> Redis 兜底 -> 过期先返旧值后台刷新）。 +- K 线层：单指数日线全量。国内指数（000001.SH 形式）复用 index_series.get_index_daily； + 国际指数走本模块 index_global 分页拉全量（单次 4000），进程内 + Redis 两级缓存。 +- 元数据：国内指数 pro.index_basic 按需拉（24h 缓存）；国际指数 tushare 无元数据， + 内置静态表（名称/地区/国家）。 +- 估值（pro.index_dailybasic，仅 8 大国内指数有数据）与成分权重（pro.index_weight， + 月度，仅国内指数）：按需拉 + Redis 中长 TTL 缓存。 +""" +from __future__ import annotations + +import asyncio +import json +import math +import time +from datetime import date, datetime, timedelta + +from .. import cache +from ..config import settings +from ..domain import Bar + +# ---- 静态元数据表（tushare index_global 支持的全部 21 个指数，展示顺序即文档顺序）---- +# region: americas 美洲 / europe 欧洲 / asia 亚太（含港股与富时A50） +GLOBAL_INDEXES: list[dict] = [ + {"code": "DJI", "name": "道琼斯工业指数", "region": "americas", "country": "美国"}, + {"code": "SPX", "name": "标普500", "region": "americas", "country": "美国"}, + {"code": "IXIC", "name": "纳斯达克综合指数", "region": "americas", "country": "美国"}, + {"code": "RUT", "name": "罗素2000", "region": "americas", "country": "美国"}, + {"code": "SPTSX", "name": "加拿大S&P/TSX", "region": "americas", "country": "加拿大"}, + {"code": "IBOVESPA", "name": "巴西IBOVESPA", "region": "americas", "country": "巴西"}, + {"code": "FTSE", "name": "富时100", "region": "europe", "country": "英国"}, + {"code": "FCHI", "name": "法国CAC40", "region": "europe", "country": "法国"}, + {"code": "GDAXI", "name": "德国DAX", "region": "europe", "country": "德国"}, + {"code": "CSX5P", "name": "STOXX欧洲50", "region": "europe", "country": "欧洲"}, + {"code": "RTS", "name": "俄罗斯RTS", "region": "europe", "country": "俄罗斯"}, + {"code": "HSI", "name": "恒生指数", "region": "asia", "country": "中国香港"}, + {"code": "HKTECH", "name": "恒生科技指数", "region": "asia", "country": "中国香港"}, + {"code": "HKAH", "name": "恒生AH股H指数", "region": "asia", "country": "中国香港"}, + {"code": "XIN9", "name": "富时中国A50", "region": "asia", "country": "新加坡"}, + {"code": "N225", "name": "日经225", "region": "asia", "country": "日本"}, + {"code": "KS11", "name": "韩国综合指数", "region": "asia", "country": "韩国"}, + {"code": "TWII", "name": "台湾加权指数", "region": "asia", "country": "中国台湾"}, + {"code": "AS51", "name": "澳大利亚标普200", "region": "asia", "country": "澳大利亚"}, + {"code": "SENSEX", "name": "印度孟买SENSEX", "region": "asia", "country": "印度"}, + {"code": "CKLSE", "name": "马来西亚指数", "region": "asia", "country": "马来西亚"}, +] +GLOBAL_META = {g["code"]: g for g in GLOBAL_INDEXES} + +# 国内指数白名单（K 线 / 详情 / 权重可用范围，防止任意 code 打爆 tushare） +CN_INDEXES: dict[str, str] = { + "000001.SH": "上证指数", + "399001.SZ": "深证成指", + "399006.SZ": "创业板指", + "000688.SH": "科创50", + "000300.SH": "沪深300", + "000016.SH": "上证50", + "000905.SH": "中证500", + "000852.SH": "中证1000", + "399016.SZ": "深证100", +} + +# index_dailybasic 实际有数据的指数（接口文档写 6 个，实测含 000300/399016 共 8 个） +DAILYBASIC_CODES = { + "000001.SH", "000016.SH", "000300.SH", "000905.SH", + "399001.SZ", "399005.SZ", "399006.SZ", "399016.SZ", +} + +_SPARK_DAYS = 45 +_HISTORY_DAYS = 150 # 日历日窗口（约 100 交易日，够取 spark） +_LIST_KEY = "global_indexes:eod:v1" +_BARS_KEY = "idxgb:" # + code（全量日线紧凑 JSON） +_BASIC_KEY = "idxbm:" # + code（index_basic 元数据） +_VAL_KEY = "idxvm:" # + code（dailybasic 估值序列） +_W_KEY = "idxwm:" # + code（index_weight 最近月度） +_PAGE = 4000 # index_global 单次返回上限 +_CALL_INTERVAL = 0.12 # 顺序调用间隔（秒），对 tushare 控频 + + +class GlobalIndexError(RuntimeError): + """全部国际指数都拉不到（token/网络故障）——接口层转 503。""" + + +def _f(v) -> float | None: + """pandas 值 -> float；NaN/None -> None。""" + if v is None: + return None + try: + f = float(v) + except (TypeError, ValueError): + return None + return None if math.isnan(f) else f + + +def _d(v) -> str | None: + """YYYYMMDD -> 'YYYY-MM-DD'（字符串便于 JSON 缓存）。""" + return datetime.strptime(str(v), "%Y%m%d").date().isoformat() if v else None + + +def is_cn_index(code: str) -> bool: + return "." in code + + +def ensure_known(code: str) -> bool: + """详情/K线/权重接口只放行白名单内的 code。""" + return code in CN_INDEXES or code in GLOBAL_META + + +# ======================= 列表层：21 个国际指数最新行情（SWR 整包） ======================= + +def _get_pro(): + from .tushare_provider import get_pro + + return get_pro() + + +def _fetch_quote_sync(pro, ts_code: str) -> dict: + """单个指数近 _HISTORY_DAYS 日行情 -> 最新一根 + spark（旧 -> 新）。""" + start = (datetime.now() - timedelta(days=_HISTORY_DAYS)).strftime("%Y%m%d") + if is_cn_index(ts_code): + df = pro.index_daily(ts_code=ts_code, start_date=start) + else: + df = pro.index_global(ts_code=ts_code, start_date=start) + if df is None or df.empty: + raise GlobalIndexError("无数据") + df = df.sort_values("trade_date") + tail = df.tail(_SPARK_DAYS) + last = df.iloc[-1] + return { + "close": _f(last["close"]), + "change": _f(last.get("change")), + "pct_chg": _f(last.get("pct_chg")), + "open": _f(last.get("open")), + "high": _f(last.get("high")), + "low": _f(last.get("low")), + "pre_close": _f(last.get("pre_close")), + "trade_date": _d(last["trade_date"]), + "spark": [round(float(c), 4) for c in tail["close"]], + "spark_dates": [str(d) for d in tail["trade_date"]], + } + + +_list_state: dict = {"payload": None} +_list_refreshing = False +_list_refresh_error: str | None = None +_bg_tasks: set[asyncio.Task] = set() + + +async def _refresh_list() -> dict: + """拉全量 21 个国际指数 EOD（顺序控频 ~8s），写进程内 state + Redis。""" + pro = await asyncio.to_thread(_get_pro) + + items: list[dict] = [] + errors: list[str] = [] + for g in GLOBAL_INDEXES: + try: + q = await asyncio.to_thread(_fetch_quote_sync, pro, g["code"]) + items.append({**g, **q}) + except Exception as e: # noqa: BLE001 —— 单指数失败不拖垮整包 + errors.append(f"{g['name']}: {str(e)[:60]}") + await asyncio.sleep(_CALL_INTERVAL) + + if not items: + raise GlobalIndexError("国际指数全部拉取失败: " + "; ".join(errors)[:200]) + + payload = { + "fetched_at": datetime.now().isoformat(), + "fetched_ts": time.time(), + "items": items, + "errors": errors, + } + _list_state["payload"] = payload + await cache.cache_set(_LIST_KEY, payload, ttl=settings.market_eod_redis_ttl) + return payload + + +async def _refresh_list_wrapped() -> None: + global _list_refresh_error, _list_refreshing + try: + await _refresh_list() + _list_refresh_error = None + except Exception as e: # noqa: BLE001 + _list_refresh_error = f"国际指数后台刷新: {str(e)[:60]}" + finally: + _list_refreshing = False + + +def _spawn_refresh() -> None: + global _list_refreshing + if _list_refreshing: + return + _list_refreshing = True + task = asyncio.create_task(_refresh_list_wrapped()) + _bg_tasks.add(task) + task.add_done_callback(_bg_tasks.discard) + + +async def fetch_global_list() -> dict: + """国际指数列表：内存新鲜直返 -> Redis 回填 -> 有旧值先返 + SWR 后台刷新 -> 冷启动同步拉。""" + p = _list_state["payload"] + if p is not None and time.time() - p["fetched_ts"] < settings.market_eod_fresh_ttl: + return p + if p is None: + cached = await cache.cache_get(_LIST_KEY) + if cached: + p = cached + _list_state["payload"] = p + if p is not None: + _spawn_refresh() + return p + return await _refresh_list() + + +async def fetch_index_quote(code: str) -> dict: + """单指数最新行情（详情页头部）。国内走 index_daily、国际走 index_global； + Redis 短缓存 2h（收盘口径一天一变）。""" + key = f"idxqt:{code}" + raw = await cache.cache_get(key) + if isinstance(raw, dict): + return raw + pro = await asyncio.to_thread(_get_pro) + q = await asyncio.to_thread(_fetch_quote_sync, pro, code) + await cache.cache_set(key, q, ttl=7200) + return q + + +# ======================= K 线层：单指数日线全量（两级缓存） ======================= + +def _fetch_global_bars_sync(ts_code: str) -> list[Bar]: + """国际指数全量日线（分页）。vol/amount 大部分指数缺失 -> volume 0 / amount None。""" + pro = _get_pro() + frames = [] + offset = 0 + while True: + df = pro.index_global(ts_code=ts_code, offset=offset, limit=_PAGE) + if df is None or df.empty: + break + frames.append(df) + if len(df) < _PAGE: + break + offset += _PAGE + time.sleep(_CALL_INTERVAL) + if not frames: + raise GlobalIndexError(f"Tushare index_global 无数据: {ts_code}") + + import pandas as pd + + df = pd.concat(frames).drop_duplicates(subset="trade_date").sort_values("trade_date") + bars: list[Bar] = [] + for _, r in df.iterrows(): + vol = _f(r.get("vol")) + amt = _f(r.get("amount")) + bars.append( + Bar( + ts=datetime.strptime(str(r["trade_date"]), "%Y%m%d"), + open=float(r["open"]), high=float(r["high"]), + low=float(r["low"]), close=float(r["close"]), + volume=vol or 0.0, + amount=amt, + ) + ) + return bars + + +# 进程内缓存（与 index_series 同款）：code -> (bars, 过期时刻) +_mem: dict[str, tuple[list[Bar], float]] = {} +_BARS_TTL = 7200 + + +def _bars_to_raw(bars: list[Bar]) -> str: + return json.dumps( + [[b.ts.isoformat(), b.open, b.high, b.low, b.close, b.volume, b.amount] for b in bars], + ensure_ascii=False, separators=(",", ":"), + ) + + +def _bars_from_raw(raw: str) -> list[Bar]: + return [ + Bar(ts=datetime.fromisoformat(row[0]), open=row[1], high=row[2], low=row[3], + close=row[4], volume=row[5], amount=row[6]) + for row in json.loads(raw) + ] + + +async def get_index_bars(code: str) -> list[Bar]: + """单指数全量日线（升序）：国内复用 index_series（上证已有热缓存），国际本模块分页。""" + if is_cn_index(code): + from .index_series import get_index_daily + + return await get_index_daily(code) + + hit = _mem.get(code) + if hit and hit[1] > time.monotonic(): + return hit[0] + key = f"{_BARS_KEY}{code}" + raw = await cache.cache_get(key) + if isinstance(raw, str): + bars = _bars_from_raw(raw) + _mem[code] = (bars, time.monotonic() + _BARS_TTL) + return bars + bars = await asyncio.to_thread(_fetch_global_bars_sync, code) + _mem[code] = (bars, time.monotonic() + _BARS_TTL) + await cache.cache_set(key, _bars_to_raw(bars), ttl=_BARS_TTL) + return bars + + +# ======================= 元数据：index_basic（国内）/ 静态表（国际） ======================= + +def _fetch_basic_sync(ts_code: str) -> dict: + df = _get_pro().index_basic(ts_code=ts_code) + if df is None or df.empty: + raise GlobalIndexError("index_basic 无数据") + r = df.iloc[-1] # 同 code 理论唯一，防御性取末行 + return { + "ts_code": str(r["ts_code"]), + "name": str(r.get("name") or ""), + "market": r.get("market"), + "publisher": r.get("publisher"), + "category": r.get("category"), + "base_date": _d(r.get("base_date")), + "base_point": _f(r.get("base_point")), + "list_date": _d(r.get("list_date")), + } + + +async def get_index_basic(code: str) -> dict | None: + """指数基本信息：国内 index_basic（24h 缓存，拉不到返 None 不阻塞）； + 国际直接由静态表合成。""" + if code in GLOBAL_META: + g = GLOBAL_META[code] + return {"ts_code": code, "name": g["name"], "market": None, "publisher": None, + "category": None, "base_date": None, "base_point": None, "list_date": None, + "country": g["country"], "region": g["region"]} + key = f"{_BASIC_KEY}{code}" + raw = await cache.cache_get(key) + if isinstance(raw, dict): + return raw + try: + basic = await asyncio.to_thread(_fetch_basic_sync, code) + except Exception: # noqa: BLE001 —— 元数据缺失时详情页行情照常 + return None + await cache.cache_set(key, basic, ttl=86400) + return basic + + +# ======================= 估值：index_dailybasic（仅部分国内指数） ======================= + +def _fetch_valuation_sync(ts_code: str, days: int) -> list[dict]: + start = (datetime.now() - timedelta(days=days)).strftime("%Y%m%d") + df = _get_pro().index_dailybasic(ts_code=ts_code, start_date=start) + if df is None or df.empty: + return [] + rows = [] + for _, r in df.sort_values("trade_date").iterrows(): + rows.append({ + "trade_date": _d(r["trade_date"]), + "pe": _f(r.get("pe")), "pe_ttm": _f(r.get("pe_ttm")), "pb": _f(r.get("pb")), + "turnover_rate": _f(r.get("turnover_rate")), + "total_mv": _f(r.get("total_mv")), "float_mv": _f(r.get("float_mv")), + }) + return rows + + +async def get_index_valuation(code: str, days: int = 400) -> list[dict]: + """近 N 日估值序列（升序）。接口只覆盖 DAILYBASIC_CODES 内的指数，其余不调接口直接空。""" + if code not in DAILYBASIC_CODES: + return [] + key = f"{_VAL_KEY}{code}:{days}" + raw = await cache.cache_get(key) + if isinstance(raw, list): + return raw + rows = await asyncio.to_thread(_fetch_valuation_sync, code, days) + await cache.cache_set(key, rows, ttl=43200) + return rows + + +# ======================= 成分权重：index_weight（月度，仅国内指数） ======================= + +def _fetch_weights_sync(ts_code: str) -> dict | None: + """index_weight 是月度快照，官方建议按整月窗口查询：本月 -> 上月 -> 前月，取首个有数据的月份。""" + pro = _get_pro() + today = date.today() + for back in range(3): + first = (today.replace(day=1) - timedelta(days=31 * back)).replace(day=1) + last_day = (first + timedelta(days=42)).replace(day=1) - timedelta(days=1) + df = pro.index_weight( + index_code=ts_code, + start_date=first.strftime("%Y%m%d"), + end_date=last_day.strftime("%Y%m%d"), + ) + if df is None or df.empty: + continue + df = df.sort_values(["trade_date", "weight"], ascending=[False, False]) + latest_date = df.iloc[0]["trade_date"] + rows = df[df["trade_date"] == latest_date] + return { + "trade_date": _d(latest_date), + "total": int(len(rows)), + "items": [ + {"con_code": str(r["con_code"]), "weight": round(float(r["weight"]), 4)} + for _, r in rows.sort_values("weight", ascending=False).iterrows() + ], + } + return None + + +async def get_index_weights(code: str) -> dict | None: + """最近月度成分权重（全量，按权重降序）。国际指数无此数据，直接 None。""" + if not is_cn_index(code): + return None + key = f"{_W_KEY}{code}" + raw = await cache.cache_get(key) + if isinstance(raw, dict): + return raw + result = await asyncio.to_thread(_fetch_weights_sync, code) + if result is None: + return None + await cache.cache_set(key, result, ttl=43200) + return result diff --git a/backend/app/data/index_series.py b/backend/app/data/index_series.py new file mode 100644 index 0000000..8a5c46b --- /dev/null +++ b/backend/app/data/index_series.py @@ -0,0 +1,96 @@ +"""A 股指数全量日线（上证指数日 K 图数据源）。 + +candles 表只有 6 位纯代码股票（TDX 导入明确排除指数，sh000001 与 sz000001 平安银行 +无法区分），指数走 tushare index_daily 按需拉取：分页拉全量（1990 年至今 ~8900 根）， +进程内 + Redis 两级缓存，历史不可变、TTL 兜到当日更新。 +""" +from __future__ import annotations + +import asyncio +import json +import time +from datetime import datetime + +from .. import cache +from ..domain import Bar + +_PAGE = 8000 # index_daily 单次返回上限 +_TTL = 7200 # 缓存 2h：历史不可变，只影响当日 bar 的新鲜度 +_CALL_GAP = 0.12 # 分页请求间隔（秒），对 tushare 控频（与 market_overview 同款） + +SH_INDEX = "000001.SH" + +# 进程内缓存：ts_code -> (bars, 过期时刻)。bars 为全量日线（升序） +_mem: dict[str, tuple[list[Bar], float]] = {} + + +def _fetch_all_sync(ts_code: str) -> list[Bar]: + """分页拉全量日线。vol 单位手 -> 股，amount 千元 -> 元（与 fetch_daily 同款换算）。""" + from .tushare_provider import get_pro + + pro = get_pro() + frames = [] + offset = 0 + while True: + df = pro.index_daily(ts_code=ts_code, offset=offset, limit=_PAGE) + if df is None or df.empty: + break + frames.append(df) + if len(df) < _PAGE: + break + offset += _PAGE + time.sleep(_CALL_GAP) + if not frames: + raise RuntimeError(f"Tushare index_daily 无数据: {ts_code}") + + import pandas as pd + + df = pd.concat(frames).drop_duplicates(subset="trade_date").sort_values("trade_date") + bars: list[Bar] = [] + for _, r in df.iterrows(): + amt = r.get("amount") + bars.append( + Bar( + ts=datetime.strptime(str(r["trade_date"]), "%Y%m%d"), + open=float(r["open"]), high=float(r["high"]), + low=float(r["low"]), close=float(r["close"]), + volume=float(r["vol"]) * 100.0, + amount=float(amt) * 1000.0 if amt is not None and amt == amt else None, + ) + ) + return bars + + +def _to_raw(bars: list[Bar]) -> str: + """紧凑 JSON：[[ts_iso, open, high, low, close, volume, amount|null], ...]（~600KB）。""" + return json.dumps( + [[b.ts.isoformat(), b.open, b.high, b.low, b.close, b.volume, b.amount] for b in bars], + ensure_ascii=False, separators=(",", ":"), + ) + + +def _from_raw(raw: str) -> list[Bar]: + return [ + Bar(ts=datetime.fromisoformat(row[0]), open=row[1], high=row[2], low=row[3], + close=row[4], volume=row[5], amount=row[6]) + for row in json.loads(raw) + ] + + +async def get_index_daily(ts_code: str = SH_INDEX) -> list[Bar]: + """全量日线（升序）：进程内 -> Redis -> tushare，未命中层级回填上一级。""" + hit = _mem.get(ts_code) + if hit and hit[1] > time.monotonic(): + return hit[0] + + key = f"idxd:{ts_code}" + raw = await cache.cache_get(key) + if isinstance(raw, str): + bars = _from_raw(raw) + _mem[ts_code] = (bars, time.monotonic() + _TTL) + return bars + + bars = await asyncio.to_thread(_fetch_all_sync, ts_code) + _mem[ts_code] = (bars, time.monotonic() + _TTL) + await cache.cache_set(key, _to_raw(bars), ttl=_TTL) + return bars diff --git a/backend/app/data/reference.py b/backend/app/data/reference.py new file mode 100644 index 0000000..6cb18f4 --- /dev/null +++ b/backend/app/data/reference.py @@ -0,0 +1,444 @@ +"""个股参考数据（tushare 参考数据版块，详情页按需单查懒加载）。 + +10 个常规 kind 走同一泛型管道：拉取 -> 精确去重 -> 按 df dtype 清洗 -> 排序 -> +5 年窗切片 -> 单行 JSON 快照 upsert（stock_reference，行即缓存，rows_json NULL=墓碑）。 + +repurchase 特殊：tushare 不支持按 ts_code 过滤（实测入参被忽略，返回全市场默认页）， +改为全市场按月分块拉取（单块触顶 2000 行则窗口减半递归）后按股拆分入库； +全局状态记在 stock_sync_state('_MARKET_', 'repurchase')——从未回填时后台任务回填 +近 24 个月（本次请求返回现有行，可能暂空），状态过期(>7 天)时内联补拉缺失月份。 + +单位沿用各接口原始口径（详见 _KINDS 注释与文档）：top10 hold_amount 股、pledge 万股、 +block_trade vol 万股 / amount 万元、repurchase vol 股 / amount 元、share_float float_share 股。 +""" +from __future__ import annotations + +import asyncio +import calendar +import json +import time +from dataclasses import dataclass +from datetime import date + +from sqlalchemy import select +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from ..config import settings +from ..db import async_session +from ..models import StockReference +from .sync_utils import call_retry, f_clean, fresh, get_pro_lazy, read_sync_state, s_clean, upsert_sync_state, utcnow + +_REFRESH_DAYS = 7 + + +@dataclass(frozen=True) +class _Kind: + api: str + fields: str # 期望列的文档化清单（实际不传 fields，见 _fetch_kind_sync 注释） + windowed: bool # True: 传 start_date(=N年前0101)+end_date(=今天) 窗口参数 + order: tuple[tuple[str, bool], ...] # 排序 (列, 是否倒序)；多列时靠稳定多趟排序 + slice_col: str # 窗口切片依据列（未来日期 >= cutoff 自然保留） + group: tuple[str, ...] | None = None # 聚合键：share_float 每股东一行，按解禁事件聚合才可展示 + years: int = 5 # 窗口年数（moneyflow 这类日频近况数据取 1 年） + refresh_days: int = 7 # 新鲜度门控（moneyflow 每日盘后更新，须 1 天） + + +def _window_start() -> str: + return f"{date.today().year - 5}0101" + + +def _window_start_for(spec: _Kind) -> str: + return f"{date.today().year - spec.years}0101" + + +def _today() -> str: + return date.today().strftime("%Y%m%d") + + +_TOP10_FIELDS = ("ts_code,ann_date,end_date,holder_name,hold_amount,hold_ratio," + "hold_float_ratio,hold_change,holder_type") +_PLEDGE_STAT_FIELDS = "ts_code,end_date,pledge_count,unrest_pledge,rest_pledge,total_share,pledge_ratio" +_PLEDGE_DETAIL_FIELDS = ("ts_code,ann_date,holder_name,pledge_amount,start_date,end_date," + "is_release,release_date,holding_amount,pledged_amount," + "p_total_ratio,h_total_ratio,pledgor,is_buyback") +_REPURCHASE_FIELDS = "ts_code,ann_date,end_date,proc,exp_date,vol,amount,high_limit,low_limit" +_FLOAT_FIELDS = "ts_code,ann_date,float_date,float_share,float_ratio,holder_name,share_type" +_BLOCK_FIELDS = "ts_code,trade_date,price,vol,amount,buyer,seller" +_HOLDERNUM_FIELDS = "ts_code,ann_date,end_date,holder_num" +_HOLDERTRADE_FIELDS = ("ts_code,ann_date,holder_name,holder_type,in_de,change_vol," + "change_ratio,after_share,after_ratio,avg_price,total_share") +_SHOCK_FIELDS = "ts_code,trade_date,name,trade_market,reason,period" +# 同花顺资金流向（万元）；2027-07-06 起源头停供 5 日主力净额与占比（net_d5_amount） +_MF_FIELDS = ("ts_code,trade_date,name,pct_change,latest,net_amount,net_d5_amount," + "buy_lg_amount,buy_lg_amount_rate,buy_md_amount,buy_md_amount_rate," + "buy_sm_amount,buy_sm_amount_rate") + +_KINDS: dict[str, _Kind] = { + # 前十大股东/流通股东：报告期窗；期内按持股数排序（tushare 返回顺序不稳定） + "top10_holders": _Kind("top10_holders", _TOP10_FIELDS, True, + (("end_date", True), ("hold_amount", True)), "end_date"), + "top10_floatholders": _Kind("top10_floatholders", _TOP10_FIELDS, True, + (("end_date", True), ("hold_amount", True)), "end_date"), + # 质押统计：周频全量（接口无窗口参数），取近 5 年切片 + "pledge_stat": _Kind("pledge_stat", _PLEDGE_STAT_FIELDS, False, + (("end_date", True),), "end_date"), + "pledge_detail": _Kind("pledge_detail", _PLEDGE_DETAIL_FIELDS, True, + (("ann_date", True),), "ann_date"), + "share_float": _Kind("share_float", _FLOAT_FIELDS, False, + (("float_date", True),), "float_date", + # 每个解禁股东一行（IPO 原始股东解禁可达数千行，接口 6000 行封顶会截断尾部）， + # 按解禁事件聚合：股数/占比求和，股东名并成「N名股东」 + group=("ann_date", "float_date", "share_type")), + "block_trade": _Kind("block_trade", _BLOCK_FIELDS, True, + (("trade_date", True),), "trade_date"), + # 同花顺资金流向：日频盘后更新，窗口 1 年、新鲜度 1 天（其余 kind 默认 5 年/7 天） + "moneyflow": _Kind("moneyflow_ths", _MF_FIELDS, True, + (("trade_date", True),), "trade_date", years=1, refresh_days=1), + "holdernumber": _Kind("stk_holdernumber", _HOLDERNUM_FIELDS, True, + (("end_date", True), ("ann_date", True)), "end_date"), + "holdertrade": _Kind("stk_holdertrade", _HOLDERTRADE_FIELDS, True, + (("ann_date", True),), "ann_date"), + "shock": _Kind("stk_shock", _SHOCK_FIELDS, True, + (("trade_date", True),), "trade_date"), + # quicksync 镜像实测无该接口数据：管道照常，墓碑优雅降级 + "high_shock": _Kind("stk_high_shock", _SHOCK_FIELDS, True, + (("trade_date", True),), "trade_date"), +} + +#: 路由白名单（repurchase 走全市场特殊管道，不在 _KINDS 泛型表里） +REFERENCE_KINDS = frozenset(_KINDS) | {"repurchase"} + + +def _kv(v): + """排序键归一：None < 字符串 < 数值，避免混合类型比较抛错。""" + if v is None: + return (0, 0.0) + if isinstance(v, str): + return (1, v) + return (2, float(v)) + + +def _clean_df(df, keep_ts_code: bool = False) -> list[dict]: + """按 df dtype 清洗（数值列 f_clean、其余列 s_clean），默认去掉冗余 ts_code 列，精确去重。 + + 数值列判定必须用 pandas.api.types.is_numeric_dtype：pandas 3 的字符串列 dtype + 是 'str' 而非 'object'，按 `!= "object"` 判断会把日期列误当数值列（f_clean 变 float）。 + """ + import pandas as pd + + df = df.drop_duplicates() + num_cols = {c for c in df.columns if pd.api.types.is_numeric_dtype(df[c])} + skip_ts = {"ts_code"} if not keep_ts_code else set() + rows: list[dict] = [] + for _, r in df.iterrows(): + row = {c: (f_clean(r[c]) if c in num_cols else s_clean(r[c])) for c in df.columns if c not in skip_ts} + if any(v is not None for v in row.values()): + rows.append(row) + return rows + + +def _group_rows(rows: list[dict], spec: _Kind) -> list[dict]: + """share_float 类聚合：数值列求和、holder_name 并成「N名股东」。""" + agg: dict[tuple, dict] = {} + order_keys: list[tuple] = [] + for r in rows: + key = tuple(r.get(c) for c in spec.group) # type: ignore[arg-type] + if key not in agg: + agg[key] = { + **{c: r.get(c) for c in spec.group}, # type: ignore[misc] + "_num": {}, "_holders": [], + } + order_keys.append(key) + a = agg[key] + for c, v in r.items(): + if isinstance(v, (int, float)) and c not in spec.group: # type: ignore[operator] + a["_num"][c] = a["_num"].get(c, 0.0) + v + h = r.get("holder_name") + if h: + a["_holders"].append(h) + out = [] + for key in order_keys: + a = agg[key] + row = {c: a[c] for c in spec.group} # type: ignore[misc] + row.update({c: round(v, 4) for c, v in a["_num"].items()}) + hs = a["_holders"] + row["holder_name"] = f"{hs[0]} 等{len(hs)}名股东" if len(hs) > 1 else (hs[0] if hs else None) + row["holder_count"] = len(hs) + out.append(row) + return out + + +def _slice_and_sort(rows: list[dict], spec: _Kind) -> list[dict]: + if spec.group: + rows = _group_rows(rows, spec) + if spec.slice_col: + cutoff = _window_start_for(spec) + rows = [r for r in rows if r.get(spec.slice_col) is None or r[spec.slice_col] >= cutoff] + # 多列排序：逆序逐列稳定排序（Python sort 稳定，后排的列为主键） + for col, desc in reversed(spec.order): + rows.sort(key=lambda r, c=col: _kv(r.get(c)), reverse=desc) + return rows + + +def _fetch_kind_sync(ts_code: str, kind: str) -> list[dict]: + """同步拉单股单分类（需在 to_thread 里跑）；无数据返回空列表。 + + 不传 fields：所用字段全为接口默认显示列，而 quicksync 镜像对部分接口 + （实测 share_float）传 fields 时会忽略 ts_code 过滤、返回全市场数据—— + 所以下面再加一道 ts_code 后过滤兜底。 + """ + spec = _KINDS[kind] + time.sleep(settings.screener_sync_interval) + params: dict = {"ts_code": ts_code} + if spec.windowed: + params["start_date"] = _window_start_for(spec) + params["end_date"] = _today() + df = call_retry(getattr(get_pro_lazy(), spec.api), **params) + if df is None or df.empty: + return [] + if "ts_code" in df.columns: + df = df[df["ts_code"] == ts_code] + return _slice_and_sort(_clean_df(df), spec) + + +# ---------- 行即缓存读写 ---------- + +def _parse(rows_json: str | None) -> list[dict]: + if not rows_json: + return [] + try: + return json.loads(rows_json) + except (ValueError, TypeError): + return [] + + +async def _read_rows(session: AsyncSession, ts_code: str, kind: str) -> list[dict]: + row = (await session.execute( + select(StockReference).where( + StockReference.ts_code == ts_code, StockReference.kind == kind + ))).scalar_one_or_none() + return _parse(row.rows_json) if row is not None else [] + + +async def _upsert_row(session: AsyncSession, ts_code: str, kind: str, rows: list[dict]) -> None: + """写单行快照（不 commit）；空列表写墓碑（rows_json NULL）。""" + stmt = pg_insert(StockReference).values( + ts_code=ts_code, kind=kind, + rows_json=json.dumps(rows, ensure_ascii=False) if rows else None, + updated_at=utcnow(), + ) + stmt = stmt.on_conflict_do_update( + index_elements=["ts_code", "kind"], + set_={"rows_json": stmt.excluded.rows_json, "updated_at": stmt.excluded.updated_at}, + ) + await session.execute(stmt) + + +# per-(code,kind) 锁：同键首次并发 N 个请求只有 1 个打 tushare，其余等锁后双检命中 +_key_locks: dict[tuple[str, str], asyncio.Lock] = {} +_guard = asyncio.Lock() + + +async def _get_generic(session: AsyncSession, ts_code: str, kind: str) -> list[dict]: + """常规 kind 的读穿透：新鲜直返；否则锁内双检 -> to_thread 拉 -> upsert。""" + spec = _KINDS[kind] + row = (await session.execute( + select(StockReference).where( + StockReference.ts_code == ts_code, StockReference.kind == kind + ))).scalar_one_or_none() + if row is not None and fresh(row.updated_at, spec.refresh_days): + return _parse(row.rows_json) + # 释放请求会话连接（close 而非 rollback：rollback 会 expire 鉴权缓存的 User，见 finance.py 注释） + await session.close() + + key = (ts_code, kind) + async with _guard: + lock = _key_locks.setdefault(key, asyncio.Lock()) + async with lock: + async with async_session() as s2: + row = (await s2.execute( + select(StockReference).where( + StockReference.ts_code == ts_code, StockReference.kind == kind + ))).scalar_one_or_none() + if row is not None and fresh(row.updated_at, spec.refresh_days): + return _parse(row.rows_json) + old = _parse(row.rows_json) if row is not None else [] + try: + rows = await asyncio.to_thread(_fetch_kind_sync, ts_code, kind) + except Exception: + if old: # 降级：返旧行，不把「上游挂了」伪装成「无数据」 + return old + raise + await _upsert_row(s2, ts_code, kind, rows) + await s2.commit() + return rows + + +# ---------- repurchase：全市场按月分块管道 ---------- + +_MARKET_CODE = "_MARKET_" +_BACKFILL_MONTHS = 24 +_REPURCHASE_CHUNK_CAP = 2000 # 接口单次上限（触顶则窗口减半递归拆分） + +_repurchase_lock = asyncio.Lock() # 回填任务与内联增量共用，防止并发重复拉 +_repurchase_task: asyncio.Task | None = None + + +def _month_range(start_ym: tuple[int, int], end_ym: tuple[int, int]) -> list[tuple[int, int]]: + """[(y, m), ...] 闭区间月列表。""" + out = [] + y, m = start_ym + while (y, m) <= end_ym: + out.append((y, m)) + m += 1 + if m > 12: + y, m = y + 1, 1 + return out + + +def _fetch_repurchase_window(pro, start: str, end: str) -> list[dict]: + """拉一个日期窗的全市场回购（同步，线程内跑）；触顶 2000 行则窗口减半递归。 + + 不传 fields：镜像实测传 fields 时 repurchase 按日期窗也返回空，去掉后正常。 + """ + time.sleep(settings.screener_sync_interval) + df = call_retry(pro.repurchase, start_date=start, end_date=end) + if df is None or df.empty: + return [] + if len(df) >= _REPURCHASE_CHUNK_CAP and start < end: + mid_ts = ( + date(int(start[:4]), int(start[4:6]), int(start[6:8])) + + (date(int(end[:4]), int(end[4:6]), int(end[6:8])) - date(int(start[:4]), int(start[4:6]), int(start[6:8]))) / 2 + ) + mid = mid_ts.strftime("%Y%m%d") + if mid > start and mid < end: + return _fetch_repurchase_window(pro, start, mid) + _fetch_repurchase_window(pro, mid, end) + return _clean_df(df, keep_ts_code=True) # 保留 ts_code 作为按股分组键 + + +def _fetch_repurchase_market(months: list[tuple[int, int]]) -> dict[str, list[dict]]: + """按月分块拉全市场回购，按股分组返回（同步，线程内跑）。 + + 注意 _clean_df 必须保留 ts_code（分组键），写入时再剥掉。 + """ + pro = get_pro_lazy() + by_code: dict[str, list[dict]] = {} + for y, m in months: + last_day = calendar.monthrange(y, m)[1] + for r in _fetch_repurchase_window(pro, f"{y}{m:02d}01", f"{y}{m:02d}{last_day}"): + code = r.get("ts_code") + if code: + by_code.setdefault(code, []).append({k: v for k, v in r.items() if k != "ts_code"}) + spec = _Kind("", _REPURCHASE_FIELDS, False, (("ann_date", True),), "ann_date") + for code in by_code: + by_code[code] = _slice_and_sort(by_code[code], spec) + return by_code + + +async def _write_repurchase_batch(session: AsyncSession, fetched: dict[str, list[dict]]) -> None: + """与库内旧行合并去重后批量 upsert（不 commit；由调用方提交）。""" + codes = list(fetched) + if not codes: + return + old_rows = (await session.execute( + select(StockReference.ts_code, StockReference.rows_json).where( + StockReference.kind == "repurchase", StockReference.ts_code.in_(codes) + ))).all() + old_map = {ts: _parse(rj) for ts, rj in old_rows} + + def _dedupe_key(r: dict) -> str: + return json.dumps(r, ensure_ascii=False, sort_keys=True) + + values = [] + for code, rows in fetched.items(): + merged = old_map.get(code, []) + seen = {_dedupe_key(r) for r in merged} + for r in rows: + k = _dedupe_key(r) + if k not in seen: + seen.add(k) + merged.append(r) + merged.sort(key=lambda r: _kv(r.get("ann_date")), reverse=True) + values.append({ + "ts_code": code, "kind": "repurchase", + "rows_json": json.dumps(merged, ensure_ascii=False), + "updated_at": utcnow(), + }) + stmt = pg_insert(StockReference).values(values) + stmt = stmt.on_conflict_do_update( + index_elements=["ts_code", "kind"], + set_={"rows_json": stmt.excluded.rows_json, "updated_at": stmt.excluded.updated_at}, + ) + await session.execute(stmt) + + +async def _sync_repurchase_locked(only_current: bool) -> dict[str, list[dict]]: + """在全局锁内拉取缺失月份并入库（调用方须已持有 _repurchase_lock）。""" + async with async_session() as s: + state = await read_sync_state(s, _MARKET_CODE, "repurchase") + now = date.today() + if state is None: + months = _month_range((now.year - _BACKFILL_MONTHS // 12, now.month), (now.year, now.month)) + else: + last = state.last_synced_at # naive UTC；增量从其所在月起补 + months = _month_range((last.year, last.month), (now.year, now.month)) + if only_current and state is not None: + # 内联增量拉近两个月（当前月 + 上月迟到公告），其余月份不会缺（首次已全量回填） + months = months[-2:] + fetched = await asyncio.to_thread(_fetch_repurchase_market, months) + await _write_repurchase_batch(s, fetched) + await upsert_sync_state(s, _MARKET_CODE, "repurchase", has_data=True) + await s.commit() + return fetched + + +async def _repurchase_backfill() -> None: + """后台全量回填（近 24 个月）；失败静默——下次触发重试。""" + try: + async with _repurchase_lock: + await _sync_repurchase_locked(only_current=False) + except Exception: # noqa: BLE001 后台任务无人接异常 + pass + + +def _spawn_repurchase_backfill() -> None: + global _repurchase_task + if _repurchase_task is not None and not _repurchase_task.done(): + return + _repurchase_task = asyncio.create_task(_repurchase_backfill()) + + +async def _get_repurchase(session: AsyncSession, ts_code: str) -> list[dict]: + state = await read_sync_state(session, _MARKET_CODE, "repurchase") + if state is not None and fresh(state.last_synced_at, _REFRESH_DAYS): + return await _read_rows(session, ts_code, "repurchase") + await session.close() # 释放请求会话连接（后面隔着网络调用） + if state is None: + # 从未回填：先返回现有行（可能空），后台任务补齐近 24 个月，下次打开即有 + _spawn_repurchase_backfill() + async with async_session() as s: + return await _read_rows(s, ts_code, "repurchase") + # 已回填但过期：内联补拉当前月（1 次调用，秒级） + async with _repurchase_lock: + async with async_session() as s: + state = await read_sync_state(s, _MARKET_CODE, "repurchase") + if state is not None and fresh(state.last_synced_at, _REFRESH_DAYS): + return await _read_rows(s, ts_code, "repurchase") + try: + await _sync_repurchase_locked(only_current=True) + except Exception: + old = await _read_rows(s, ts_code, "repurchase") + if old: + return old # 降级返旧行 + raise + return await _read_rows(s, ts_code, "repurchase") + + +async def get_reference(session: AsyncSession, ts_code: str, kind: str) -> list[dict]: + """读穿透入口。返回空列表 = 确认无数据（墓碑）；抛异常 = 上游失败且无旧行可降级。""" + if kind == "repurchase": + return await _get_repurchase(session, ts_code) + if kind not in _KINDS: + raise ValueError(f"未知参考数据分类: {kind}") + return await _get_generic(session, ts_code, kind) diff --git a/backend/app/data/sync_utils.py b/backend/app/data/sync_utils.py new file mode 100644 index 0000000..4223643 --- /dev/null +++ b/backend/app/data/sync_utils.py @@ -0,0 +1,87 @@ +"""懒加载管道共用件：tushare pro 单例、限频重试、取值清洗、stock_sync_state 读写。 + +finance / dividend 等按需单查管道共用；语义与 company.py 内的私有版本一致 +（那处历史代码未迁移，新管道一律从这里取）。 +""" +from __future__ import annotations + +import time +from datetime import datetime, timedelta, timezone + +from sqlalchemy import select +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from ..config import settings +from ..models import StockSyncState + +# 频率超限特征（等待 62s 重试一次；与 company.py / data.etf_sync._call_retry 同款语义） +_RATE_MARKS = ("频率超限", "每分钟") + +_pro = None # 惰性单例（get_pro 每次都 ts.set_token 写文件，没必要重复） + + +def get_pro_lazy(): + if not settings.tushare_token: + raise RuntimeError("未配置 TUSHARE_TOKEN，无法拉取 tushare 数据（backend/.env）") + global _pro + if _pro is None: + from .tushare_provider import get_pro + + _pro = get_pro() + return _pro + + +def call_retry(fn, *args, **kwargs): + """同步调用 tushare 接口；「每分钟」级频率超限等 62s 重试一次。""" + try: + return fn(*args, **kwargs) + except Exception as e: # noqa: BLE001 + msg = str(e) + if any(m in msg for m in _RATE_MARKS) and "小时" not in msg: + time.sleep(62) + return fn(*args, **kwargs) + raise + + +def utcnow() -> datetime: + return datetime.now(timezone.utc).replace(tzinfo=None) + + +def fresh(updated_at: datetime | None, days: int) -> bool: + return updated_at is not None and updated_at >= utcnow() - timedelta(days=days) + + +def s_clean(v) -> str | None: + """pandas NaN / 空串 / None -> None，其余 strip。""" + if v is None or (isinstance(v, float) and v != v): + return None + s = str(v).strip() + return s or None + + +def f_clean(v) -> float | None: + try: + f = float(v) + except (TypeError, ValueError): + return None + return None if f != f else f # NaN -> None + + +async def read_sync_state(session: AsyncSession, ts_code: str, kind: str) -> StockSyncState | None: + return (await session.execute( + select(StockSyncState).where( + StockSyncState.ts_code == ts_code, StockSyncState.kind == kind + ))).scalar_one_or_none() + + +async def upsert_sync_state(session: AsyncSession, ts_code: str, kind: str, *, has_data: bool) -> None: + """写入同步状态（不 commit，由调用方统一提交）。""" + stmt = pg_insert(StockSyncState).values( + ts_code=ts_code, kind=kind, last_synced_at=utcnow(), has_data=has_data, + ) + stmt = stmt.on_conflict_do_update( + index_elements=["ts_code", "kind"], + set_={"last_synced_at": stmt.excluded.last_synced_at, "has_data": stmt.excluded.has_data}, + ) + await session.execute(stmt) diff --git a/backend_run.log b/backend_run.log index b246ee1..6f3dc31 100644 --- a/backend_run.log +++ b/backend_run.log @@ -781,3 +781,82 @@ INFO: Application startup complete. INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit) INFO: 127.0.0.1:62984 - "GET /api/stocks/000001.SZ/reference/moneyflow HTTP/1.1" 200 OK INFO: 127.0.0.1:62991 - "GET /api/stocks/000001.SZ/reference/moneyflow HTTP/1.1" 200 OK +INFO: 127.0.0.1:63245 - "GET /api/auth/me HTTP/1.1" 200 OK +INFO: 127.0.0.1:63246 - "GET /api/preferences HTTP/1.1" 200 OK +INFO: 127.0.0.1:63250 - "GET /api/stocks/facets HTTP/1.1" 200 OK +INFO: 127.0.0.1:63249 - "GET /api/stocks?sort=symbol&order=asc&limit=100&offset=0 HTTP/1.1" 200 OK +INFO: 127.0.0.1:63260 - "GET /api/watchlist HTTP/1.1" 200 OK +INFO: 127.0.0.1:63256 - "GET /api/trades?ts_code=000001.SZ HTTP/1.1" 200 OK +INFO: 127.0.0.1:63258 - "GET /api/stocks/000001.SZ/dividends HTTP/1.1" 200 OK +INFO: 127.0.0.1:63259 - "GET /api/screener/preview/000001.SZ?limit=500&adjust=qfq&timeframe=1d HTTP/1.1" 200 OK +INFO: 127.0.0.1:63276 - "GET /api/stocks/000001.SZ/company HTTP/1.1" 200 OK +INFO: 127.0.0.1:63275 - "GET /api/stocks/000001.SZ/finance HTTP/1.1" 200 OK +INFO: 127.0.0.1:63279 - "GET /api/stocks/000001.SZ/reference/top10_holders HTTP/1.1" 200 OK +INFO: 127.0.0.1:63272 - "GET /api/screener/preview/000001.SZ?limit=800&adjust=qfq&timeframe=1d&end=2024-08-14 HTTP/1.1" 200 OK +INFO: 127.0.0.1:63281 - "GET /api/stocks/000001.SZ/reference/moneyflow HTTP/1.1" 200 OK +INFO: 127.0.0.1:11920 - "GET /api/auth/me HTTP/1.1" 200 OK +INFO: 127.0.0.1:11937 - "GET /api/stocks/facets HTTP/1.1" 200 OK +INFO: 127.0.0.1:11933 - "GET /api/preferences HTTP/1.1" 200 OK +INFO: 127.0.0.1:11936 - "GET /api/stocks?search=%E5%8D%AB%E6%98%9F%E5%8C%96%E5%AD%A6&sort=symbol&order=asc&limit=100&offset=0 HTTP/1.1" 200 OK +INFO: 127.0.0.1:11942 - "GET /api/stocks/002648.SZ/dividends HTTP/1.1" 200 OK +INFO: 127.0.0.1:11941 - "GET /api/trades?ts_code=002648.SZ HTTP/1.1" 200 OK +INFO: 127.0.0.1:11943 - "GET /api/screener/preview/002648.SZ?limit=500&adjust=qfq&timeframe=1d HTTP/1.1" 200 OK +INFO: 127.0.0.1:11949 - "GET /api/watchlist HTTP/1.1" 200 OK +INFO: 127.0.0.1:11967 - "GET /api/stocks/002648.SZ/company HTTP/1.1" 200 OK +INFO: 127.0.0.1:11964 - "GET /api/stocks/002648.SZ/finance HTTP/1.1" 200 OK +INFO: 127.0.0.1:11963 - "GET /api/screener/preview/002648.SZ?limit=800&adjust=qfq&timeframe=1d&end=2024-08-14 HTTP/1.1" 200 OK +INFO: 127.0.0.1:12029 - "GET /api/stocks/002648.SZ/reference/top10_holders HTTP/1.1" 200 OK +INFO: 127.0.0.1:12088 - "GET /api/market/overview HTTP/1.1" 200 OK +INFO: 127.0.0.1:12093 - "GET /api/screener/sync/status HTTP/1.1" 200 OK +INFO: 127.0.0.1:12094 - "GET /api/etf/sync/status HTTP/1.1" 200 OK +INFO: 127.0.0.1:12139 - "GET /api/market/global-indexes HTTP/1.1" 200 OK +INFO: 127.0.0.1:12155 - "GET /api/market/indexes/DJI HTTP/1.1" 200 OK +INFO: 127.0.0.1:12160 - "GET /api/market/indexes/DJI/candles?timeframe=1d HTTP/1.1" 200 OK +INFO: 127.0.0.1:12223 - "GET /api/market/global-indexes HTTP/1.1" 200 OK +INFO: 127.0.0.1:12288 - "GET /api/market/overview HTTP/1.1" 200 OK +INFO: 127.0.0.1:12293 - "GET /api/screener/sync/status HTTP/1.1" 200 OK +INFO: 127.0.0.1:12294 - "GET /api/etf/sync/status HTTP/1.1" 200 OK +INFO: 127.0.0.1:12303 - "GET /api/stocks/facets HTTP/1.1" 200 OK +INFO: 127.0.0.1:12302 - "GET /api/stocks?sort=symbol&order=asc&limit=100&offset=0 HTTP/1.1" 200 OK +INFO: 127.0.0.1:12309 - "GET /api/stocks?watched_only=true&sort=symbol&order=asc&limit=100&offset=0 HTTP/1.1" 200 OK +INFO: 127.0.0.1:12314 - "GET /api/trades?ts_code=002100.SZ HTTP/1.1" 200 OK +INFO: 127.0.0.1:12326 - "GET /api/watchlist HTTP/1.1" 200 OK +INFO: 127.0.0.1:12325 - "GET /api/screener/preview/002100.SZ?limit=500&adjust=qfq&timeframe=1d HTTP/1.1" 200 OK +INFO: 127.0.0.1:12324 - "GET /api/stocks/002100.SZ/dividends HTTP/1.1" 200 OK +INFO: 127.0.0.1:12333 - "GET /api/screener/preview/002100.SZ?limit=800&adjust=qfq&timeframe=1d&end=2024-08-14 HTTP/1.1" 200 OK +INFO: 127.0.0.1:12335 - "GET /api/stocks/002100.SZ/company HTTP/1.1" 200 OK +INFO: 127.0.0.1:12334 - "GET /api/stocks/002100.SZ/finance HTTP/1.1" 200 OK +INFO: 127.0.0.1:12344 - "GET /api/stocks/002100.SZ/reference/top10_holders HTTP/1.1" 200 OK +INFO: 127.0.0.1:12353 - "GET /api/stocks/002100.SZ/reference/block_trade HTTP/1.1" 200 OK +INFO: 127.0.0.1:12528 - "GET /api/stocks/002100.SZ/reference/holdertrade HTTP/1.1" 200 OK +INFO: 127.0.0.1:12543 - "GET /api/stocks/002100.SZ/reference/holdernumber HTTP/1.1" 200 OK +INFO: 127.0.0.1:12573 - "GET /api/auth/me HTTP/1.1" 200 OK +INFO: 127.0.0.1:12580 - "GET /api/stocks/facets HTTP/1.1" 200 OK +INFO: 127.0.0.1:12577 - "GET /api/preferences HTTP/1.1" 200 OK +INFO: 127.0.0.1:12579 - "GET /api/stocks?sort=symbol&order=asc&limit=100&offset=0 HTTP/1.1" 200 OK +INFO: 127.0.0.1:12585 - "GET /api/watchlist HTTP/1.1" 200 OK +INFO: 127.0.0.1:12582 - "GET /api/trades?ts_code=000001.SZ HTTP/1.1" 200 OK +INFO: 127.0.0.1:12583 - "GET /api/stocks/000001.SZ/dividends HTTP/1.1" 200 OK +INFO: 127.0.0.1:12584 - "GET /api/screener/preview/000001.SZ?limit=500&adjust=qfq&timeframe=1d HTTP/1.1" 200 OK +INFO: 127.0.0.1:12595 - "GET /api/stocks/000001.SZ/company HTTP/1.1" 200 OK +INFO: 127.0.0.1:12594 - "GET /api/stocks/000001.SZ/finance HTTP/1.1" 200 OK +INFO: 127.0.0.1:12591 - "GET /api/screener/preview/000001.SZ?limit=800&adjust=qfq&timeframe=1d&end=2024-08-14 HTTP/1.1" 200 OK +INFO: 127.0.0.1:12599 - "GET /api/stocks/000001.SZ/reference/top10_holders HTTP/1.1" 200 OK +INFO: 127.0.0.1:12604 - "GET /api/stocks/000001.SZ/reference/moneyflow HTTP/1.1" 200 OK +INFO: 127.0.0.1:12607 - "GET /api/stocks/002100.SZ/reference/moneyflow HTTP/1.1" 200 OK +INFO: 127.0.0.1:12676 - "GET /api/screener/preview/002100.SZ?limit=800&adjust=qfq&timeframe=1d&end=2021-04-28 HTTP/1.1" 200 OK +INFO: 127.0.0.1:12826 - "GET /api/screener/preview/002100.SZ?limit=800&adjust=qfq&timeframe=1d&end=2018-01-10 HTTP/1.1" 200 OK +INFO: 127.0.0.1:12831 - "GET /api/screener/preview/002100.SZ?limit=500&adjust=qfq&timeframe=1w HTTP/1.1" 200 OK +INFO: 127.0.0.1:12833 - "PUT /api/preferences HTTP/1.1" 200 OK +INFO: 127.0.0.1:12841 - "GET /api/stocks/002100.SZ/finance HTTP/1.1" 200 OK +INFO: 127.0.0.1:12842 - "GET /api/stocks/002100.SZ/company HTTP/1.1" 200 OK +INFO: 127.0.0.1:12838 - "GET /api/screener/preview/002100.SZ?limit=800&adjust=qfq&timeframe=1w&end=2017-01-16 HTTP/1.1" 200 OK +INFO: 127.0.0.1:12880 - "GET /api/screener/preview/002100.SZ?limit=500&adjust=qfq&timeframe=1d HTTP/1.1" 200 OK +INFO: 127.0.0.1:12886 - "GET /api/screener/preview/002100.SZ?limit=800&adjust=qfq&timeframe=1d&end=2024-08-14 HTTP/1.1" 200 OK +INFO: 127.0.0.1:12890 - "GET /api/stocks/002100.SZ/finance HTTP/1.1" 200 OK +INFO: 127.0.0.1:12891 - "GET /api/stocks/002100.SZ/company HTTP/1.1" 200 OK +INFO: 127.0.0.1:12892 - "PUT /api/preferences HTTP/1.1" 200 OK +INFO: 127.0.0.1:12896 - "GET /api/screener/preview/002100.SZ?limit=800&adjust=qfq&timeframe=1d&end=2021-04-28 HTTP/1.1" 200 OK +INFO: 127.0.0.1:12901 - "GET /api/screener/preview/002100.SZ?limit=800&adjust=qfq&timeframe=1d&end=2018-01-10 HTTP/1.1" 200 OK +INFO: 127.0.0.1:12942 - "GET /api/screener/preview/002100.SZ?limit=800&adjust=qfq&timeframe=1d&end=2024-08-14 HTTP/1.1" 200 OK +INFO: 127.0.0.1:12944 - "PUT /api/preferences HTTP/1.1" 200 OK diff --git a/frontend/src/chartStyles.ts b/frontend/src/chartStyles.ts new file mode 100644 index 0000000..4e1cc06 --- /dev/null +++ b/frontend/src/chartStyles.ts @@ -0,0 +1,32 @@ +// klinecharts v10 暗色主题（详情页与首页指数图共用）。 +// UP/DOWN 由调用方传入（跟随设置的涨跌配色），bars[0] 与库默认项深合并， +// 只覆盖涨跌色等少量键——init 时 merge 到 getDefaultStyles。 +export function darkStyles(up: string, down: string) { + return { + grid: { horizontal: { color: '#1C1E24' }, vertical: { color: '#1C1E24' } }, + // 内建 VOL 等指标柱的涨跌色缺省是库默认绿涨红跌，与全站语义相反 + indicator: { bars: [{ upColor: up, downColor: down, noChangeColor: '#7A818C' }] }, + candle: { + bar: { + upColor: up, downColor: down, + upBorderColor: up, downBorderColor: down, + upWickColor: up, downWickColor: down, + }, + priceMark: { + high: { color: '#9AA0AA' }, low: { color: '#9AA0AA' }, + last: { upColor: up, downColor: down }, + }, + }, + xAxis: { axisLine: { color: '#2A2D34' }, tickText: { color: '#9AA0AA', size: 12 }, tickLine: { color: '#2A2D34' } }, + yAxis: { axisLine: { color: '#2A2D34' }, tickText: { color: '#9AA0AA', size: 12 }, tickLine: { color: '#2A2D34' } }, + crosshair: { + horizontal: { text: { backgroundColor: '#333A45' } }, + vertical: { text: { backgroundColor: '#333A45' } }, + }, + separator: { + color: '#23252B', + // 悬停/拖拽分隔条时的底色（库默认 8% 蓝在纯黑底上不可见，加重为可感知的拖拽提示） + activeBackgroundColor: 'rgba(37, 99, 235, 0.30)', + }, + }; +} diff --git a/frontend/src/components/AmountHistoryChart.vue b/frontend/src/components/AmountHistoryChart.vue new file mode 100644 index 0000000..ba0523b --- /dev/null +++ b/frontend/src/components/AmountHistoryChart.vue @@ -0,0 +1,133 @@ + + + diff --git a/frontend/src/components/CompanyInfoPanel.vue b/frontend/src/components/CompanyInfoPanel.vue new file mode 100644 index 0000000..c8849e8 --- /dev/null +++ b/frontend/src/components/CompanyInfoPanel.vue @@ -0,0 +1,130 @@ + + + diff --git a/frontend/src/components/EtfSyncBar.vue b/frontend/src/components/EtfSyncBar.vue new file mode 100644 index 0000000..0d4ce4e --- /dev/null +++ b/frontend/src/components/EtfSyncBar.vue @@ -0,0 +1,75 @@ + + + diff --git a/frontend/src/components/FinancePanel.vue b/frontend/src/components/FinancePanel.vue new file mode 100644 index 0000000..b044666 --- /dev/null +++ b/frontend/src/components/FinancePanel.vue @@ -0,0 +1,167 @@ + + + diff --git a/frontend/src/components/IndexKLine.vue b/frontend/src/components/IndexKLine.vue new file mode 100644 index 0000000..9b8c9d8 --- /dev/null +++ b/frontend/src/components/IndexKLine.vue @@ -0,0 +1,133 @@ + + + diff --git a/frontend/src/components/ReferencePanel.vue b/frontend/src/components/ReferencePanel.vue new file mode 100644 index 0000000..a816ba8 --- /dev/null +++ b/frontend/src/components/ReferencePanel.vue @@ -0,0 +1,467 @@ + + + diff --git a/frontend/src/components/Sparkline.vue b/frontend/src/components/Sparkline.vue new file mode 100644 index 0000000..c5bfbf2 --- /dev/null +++ b/frontend/src/components/Sparkline.vue @@ -0,0 +1,90 @@ + + + diff --git a/frontend/src/stores/etfSync.ts b/frontend/src/stores/etfSync.ts new file mode 100644 index 0000000..686c1fd --- /dev/null +++ b/frontend/src/stores/etfSync.ts @@ -0,0 +1,53 @@ +import { defineStore } from 'pinia'; +import { ref } from 'vue'; +import { getEtfSyncStatus, startEtfSync } from '@/api/client'; +import type { EtfSyncStatus } from '@/api/types'; + +/** 全市场 ETF 同步（ETF 页手动触发）。与 A 股同步 store 解耦，独立维护轮询。 */ +export const useEtfSyncStore = defineStore('etfSync', () => { + const syncStatus = ref(null); + const error = ref(null); // 启动同步的请求级错误 + let pollTimer: ReturnType | null = null; + + async function fetchStatus() { + try { + syncStatus.value = await getEtfSyncStatus(); + } catch { + /* 静默：状态拉取失败不阻塞页面 */ + } + } + + function stopPolling() { + if (pollTimer) { + clearInterval(pollTimer); + pollTimer = null; + } + } + + /** 正在同步时开始 2s 轮询；空闲则停轮询（进入页面时也调用，承接后台遗留任务）。 */ + function pollIfRunning() { + if (syncStatus.value?.running) { + if (!pollTimer) { + pollTimer = setInterval(async () => { + await fetchStatus(); + if (!syncStatus.value?.running) stopPolling(); + }, 2000); + } + } else { + stopPolling(); + } + } + + async function startSync(full = false) { + error.value = null; + try { + await startEtfSync(full); + await fetchStatus(); + pollIfRunning(); + } catch (e) { + error.value = e instanceof Error ? e.message : '启动同步失败'; + } + } + + return { syncStatus, error, fetchStatus, startSync, stopPolling, pollIfRunning }; +}); diff --git a/frontend/src/views/EtfsView.vue b/frontend/src/views/EtfsView.vue new file mode 100644 index 0000000..d641da8 --- /dev/null +++ b/frontend/src/views/EtfsView.vue @@ -0,0 +1,391 @@ + + + diff --git a/frontend/src/views/IndexDetailView.vue b/frontend/src/views/IndexDetailView.vue new file mode 100644 index 0000000..7e830ed --- /dev/null +++ b/frontend/src/views/IndexDetailView.vue @@ -0,0 +1,367 @@ + + + diff --git a/frontend/src/views/IndexesView.vue b/frontend/src/views/IndexesView.vue new file mode 100644 index 0000000..9f1c28c --- /dev/null +++ b/frontend/src/views/IndexesView.vue @@ -0,0 +1,169 @@ + + + diff --git a/frontend_dev.log b/frontend_dev.log index 1036c6d..5db3128 100644 --- a/frontend_dev.log +++ b/frontend_dev.log @@ -9,3 +9,54 @@ $ vite 18:31:31 [vite] (client) page reload src/api/client.ts 18:31:45 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\StockDetailOverlay.vue, /src/style.css 18:31:48 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\StockDetailOverlay.vue, /src/style.css +19:07:53 [vite] (client) page reload src/api/types.ts +19:07:58 [vite] (client) page reload src/api/client.ts +19:08:04 [vite] (client) page reload src/api/client.ts +19:08:48 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\MarketOverview.vue, /src/style.css +19:08:54 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\MarketOverview.vue, /src/style.css +19:09:01 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\MarketOverview.vue, /src/style.css +19:09:07 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\MarketOverview.vue, /src/style.css +19:09:11 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\MarketOverview.vue, /src/style.css +19:11:04 [vite] (client) page reload src/router.ts +19:29:33 [vite] (client) hmr update /src/views/IndexDetailView.vue, /src/style.css +19:29:35 [vite] (client) hmr update /src/views/IndexDetailView.vue, /src/style.css +01:50:18 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\MarketOverview.vue, /src/style.css +01:50:23 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\MarketOverview.vue, /src/style.css +02:16:32 [vite] (client) page reload src/api/types.ts +02:16:33 [vite] (client) page reload src/api/types.ts +02:16:36 [vite] (client) page reload src/api/client.ts +02:16:46 [vite] (client) page reload src/api/client.ts +02:16:47 [vite] (client) page reload src/stores/settings.ts +02:16:48 [vite] (client) page reload src/stores/settings.ts +02:17:18 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\DetailKLine.vue, /src/style.css +02:17:20 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\DetailKLine.vue, /src/style.css +02:17:40 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\DetailKLine.vue, /src/style.css +02:17:41 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\DetailKLine.vue, /src/style.css +02:17:42 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\DetailKLine.vue, /src/style.css +02:17:43 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\DetailKLine.vue, /src/style.css +02:17:43 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\DetailKLine.vue, /src/style.css +02:17:44 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\DetailKLine.vue, /src/style.css +02:18:28 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\StockDetailOverlay.vue, /src/style.css +02:18:30 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\StockDetailOverlay.vue, /src/style.css +02:18:35 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\StockDetailOverlay.vue, /src/style.css +02:18:49 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\StockDetailOverlay.vue, /src/style.css +02:18:55 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\StockDetailOverlay.vue, /src/style.css +02:18:57 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\StockDetailOverlay.vue, /src/style.css +02:18:57 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\StockDetailOverlay.vue, /src/style.css +02:19:09 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\StockDetailOverlay.vue, /src/style.css +02:19:09 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\StockDetailOverlay.vue, /src/style.css +02:24:43 [vite] http proxy error: /api/auth/me +AggregateError [ECONNREFUSED]: + at internalConnectMultiple (node:net:1134:18) + at afterConnectMultiple (node:net:1715:7) +07:54:29 [vite] (client) page reload src/api/types.ts +07:54:32 [vite] (client) page reload src/api/client.ts +07:54:39 [vite] (client) page reload src/api/client.ts +07:55:37 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\StockDetailOverlay.vue, /src/style.css +07:55:38 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\StockDetailOverlay.vue, /src/style.css +08:25:59 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\ReferencePanel.vue, /src/style.css +12:48:22 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\ReferencePanel.vue, /src/style.css +12:48:23 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\ReferencePanel.vue, /src/style.css +12:56:52 [vite] (client) hmr update /src/style.css, /@fs/D:\Project\stock\frontend\src\components\ReferencePanel.vue, /@fs/D:\Project\stock\frontend\src\components\ConditionChips.vue +12:56:59 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\ReferencePanel.vue, /src/style.css +13:30:58 [vite] (client) hmr update /@fs/D:\Project\stock\frontend\src\components\StockDetailOverlay.vue, /src/style.css