diff --git a/README.md b/README.md index 85e4f84..cc37a6a 100644 --- a/README.md +++ b/README.md @@ -218,6 +218,7 @@ PORT=3018 # 服务端口 | [docs/deployment.md](./docs/deployment.md) | 部署方式(Dev / Docker / GH Actions)、老 CPU 兼容、更新代码、访问密码 | | [docs/configuration.md](./docs/configuration.md) | 所有 `.env` 配置项详解(数据源、AI、服务、密码、数据目录) | | [docs/features.md](./docs/features.md) | 各功能模块详细说明(选股/指标/回测/监控/个股分析/数据扩展) | +| [docs/custom-data-source.md](./docs/custom-data-source.md) | 自定义数据源接入、YAML 配置与 mock 联调示例 | | [docs/strategy.md](./docs/strategy.md) | 策略体系(18 内置策略 + 三种扩展方式 + 文件结构) | | [backend/app/strategy/prompts/strategy-guide.md](./backend/app/strategy/prompts/strategy-guide.md) | 策略开发完整规范(AI 生成与手写) | diff --git a/backend/app/api/alerts.py b/backend/app/api/alerts.py index fe0e4d4..87c1922 100644 --- a/backend/app/api/alerts.py +++ b/backend/app/api/alerts.py @@ -136,9 +136,7 @@ def seed_demo_alerts(request: Request, count: int = 12, recent: bool = True): "signals": ev["signals"], "severity": ev.get("severity", "info"), } for ev in events] - with qs._lock: - qs._pending_alerts.extend(sse_alerts) - qs._alert_event.set() + qs.push_alerts(sse_alerts) return {"ok": True, "generated": len(events)} diff --git a/backend/app/api/data.py b/backend/app/api/data.py index 8c69c5b..08cee45 100644 --- a/backend/app/api/data.py +++ b/backend/app/api/data.py @@ -663,8 +663,7 @@ def clear_data(request: Request): # - 待推送的实时通知队列 (进程内存) qs = getattr(request.app.state, "quote_service", None) if qs is not None: - with qs._lock: - qs._pending_alerts.clear() + qs.clear_pending_alerts() # 清除 Polars 缓存 # 先 clear_cache 无条件清空内存 (refresh_cache 在磁盘无数据时会提前 return, diff --git a/backend/app/api/financials.py b/backend/app/api/financials.py index b8414b3..b69d00d 100644 --- a/backend/app/api/financials.py +++ b/backend/app/api/financials.py @@ -18,11 +18,26 @@ logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/financials", tags=["financials"]) +def _financial_allowed(capset) -> bool: + """是否有财务数据访问权限 (TickFlow FINANCIAL 套餐 或 custom 财务源)。""" + if capset.has(Cap.FINANCIAL): + return True + from app.services.financial_sync import _financial_is_custom + return _financial_is_custom() + + +def _require_financial(capset) -> None: + """_require_financial(capset) 的 custom 感知版本。""" + if not _financial_allowed(capset): + from app.tickflow.capabilities import CapabilityDenied + raise CapabilityDenied(Cap.FINANCIAL) + + @router.get("/status") def financial_status(request: Request): """返回各财务表的同步状态。无需 FINANCIAL 权限（前端根据 available 决定是否展示）。""" capset = request.app.state.capabilities - if not capset.has(Cap.FINANCIAL): + if not _financial_allowed(capset): return {"available": False, "tables": {}} data_dir = request.app.state.repo.store.data_dir @@ -59,7 +74,7 @@ def financial_status(request: Request): def get_metrics(request: Request, symbol: str | None = None): """查询核心财务指标。""" capset = request.app.state.capabilities - capset.require(Cap.FINANCIAL) + _require_financial(capset) df = get_financial_df(request.app.state.repo.store.data_dir, "metrics") if df.is_empty(): @@ -73,7 +88,7 @@ def get_metrics(request: Request, symbol: str | None = None): def get_income(request: Request, symbol: str | None = None): """查询利润表。""" capset = request.app.state.capabilities - capset.require(Cap.FINANCIAL) + _require_financial(capset) df = get_financial_df(request.app.state.repo.store.data_dir, "income") if df.is_empty(): @@ -87,7 +102,7 @@ def get_income(request: Request, symbol: str | None = None): def get_balance_sheet(request: Request, symbol: str | None = None): """查询资产负债表。""" capset = request.app.state.capabilities - capset.require(Cap.FINANCIAL) + _require_financial(capset) df = get_financial_df(request.app.state.repo.store.data_dir, "balance_sheet") if df.is_empty(): @@ -101,7 +116,7 @@ def get_balance_sheet(request: Request, symbol: str | None = None): def get_cash_flow(request: Request, symbol: str | None = None): """查询现金流量表。""" capset = request.app.state.capabilities - capset.require(Cap.FINANCIAL) + _require_financial(capset) df = get_financial_df(request.app.state.repo.store.data_dir, "cash_flow") if df.is_empty(): @@ -120,7 +135,7 @@ def sync_table(request: Request, table: str): 前端通过轮询 GET /status 的 syncing 字段观察进度。 """ capset = request.app.state.capabilities - capset.require(Cap.FINANCIAL) + _require_financial(capset) valid_tables = {"metrics", "income", "balance_sheet", "cash_flow", "all"} if table not in valid_tables: @@ -151,7 +166,7 @@ async def analyze_financials(request: Request, req: AnalyzeRequest): 以便前端用 ReadableStream 逐行解析,更简单可靠)。 """ capset = request.app.state.capabilities - capset.require(Cap.FINANCIAL) + _require_financial(capset) if not req.symbol: raise HTTPException(400, "symbol 不能为空") @@ -187,7 +202,7 @@ class SaveReportRequest(BaseModel): def list_reports(request: Request): """获取全部历史报告(按时间降序,后端已裁剪到上限)。无需 FINANCIAL 能力读取列表元信息。""" capset = request.app.state.capabilities - if not capset.has(Cap.FINANCIAL): + if not _financial_allowed(capset): return {"reports": []} return {"reports": ai_reports.list_reports()} @@ -196,7 +211,7 @@ def list_reports(request: Request): def save_report(request: Request, req: SaveReportRequest): """保存一条报告。""" capset = request.app.state.capabilities - capset.require(Cap.FINANCIAL) + _require_financial(capset) report = ai_reports.save_report({ "symbol": req.symbol, "name": req.name, @@ -212,6 +227,6 @@ def save_report(request: Request, req: SaveReportRequest): def delete_report(request: Request, report_id: str): """删除一条报告。""" capset = request.app.state.capabilities - capset.require(Cap.FINANCIAL) + _require_financial(capset) ok = ai_reports.delete_report(report_id) return {"ok": ok} diff --git a/backend/app/api/kline.py b/backend/app/api/kline.py index 8e0261f..3d3da53 100644 --- a/backend/app/api/kline.py +++ b/backend/app/api/kline.py @@ -15,6 +15,19 @@ logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/kline", tags=["kline"]) +def _minute_allowed(capset) -> bool: + """是否有分钟K权限 (TickFlow Pro+ 或 custom minute 源)。""" + from app.tickflow.capabilities import Cap + if capset.has(Cap.KLINE_MINUTE_BATCH): + return True + from app.services import preferences + provider = preferences.get_minute_data_provider() + if provider == "tickflow": + return False + from app.data_providers import custom as custom_sources + return custom_sources.provider_has_dataset(provider, "minute") + + @router.get("/instruments/search") def search_instruments( request: Request, @@ -439,7 +452,7 @@ async def sync_minute(request: Request): repo = request.app.state.repo capset = request.app.state.capabilities - if not capset.has(Cap.KLINE_MINUTE_BATCH): + if not _minute_allowed(capset): raise HTTPException(status_code=403, detail="需要 Pro+ 权限") job_id = job_store.create() @@ -663,7 +676,7 @@ async def extend_minute_history(request: Request): capset = request.app.state.capabilities from app.tickflow.capabilities import Cap - if not capset.has(Cap.KLINE_MINUTE_BATCH): + if not _minute_allowed(capset): raise HTTPException(status_code=403, detail="需要 Pro+ 权限 (batch minute K-line)") # month 单位(按月扩展更长的分钟K历史)仅 Expert+ 开放;Pro 仅可用 day @@ -727,10 +740,15 @@ async def extend_minute_history(request: Request): progress("extend_minute", 8, f"标的池: {len(universe)} 只") from app.tickflow.capabilities import Cap + from app.tickflow.rate_limits import resolve_limit - lim = capset.limits(Cap.KLINE_MINUTE_BATCH) - batch_size = lim.batch if lim and lim.batch else 100 - rpm = lim.rpm if lim else 30 + limit = resolve_limit( + capset, + Cap.KLINE_MINUTE_BATCH, + default_batch=100, + default_rpm=30, + default_rpm_when_unset=False, + ) def _run(): """全部在 executor 线程里完成,避免阻塞事件循环。""" @@ -745,7 +763,7 @@ async def extend_minute_history(request: Request): universe, start_time=_dt.combine(new_start, _dt.min.time()), end_time=_dt.combine(latest, _dt.min.time()), - batch_size=batch_size, rpm=rpm, + batch_size=limit.batch, rpm=limit.rpm, on_chunk_done=_chunk, ) diff --git a/backend/app/api/monitor_rules.py b/backend/app/api/monitor_rules.py index b5e332c..9acac34 100644 --- a/backend/app/api/monitor_rules.py +++ b/backend/app/api/monitor_rules.py @@ -479,9 +479,7 @@ def trigger_ladder(request: Request): "conditions": ev["conditions"], "logic": ev["logic"], } for ev in rule_events] try: - with quote_svc._lock: - quote_svc._pending_alerts.extend(sse_alerts) - quote_svc._alert_event.set() + quote_svc.push_alerts(sse_alerts) except Exception: # noqa: BLE001 pass diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py index 39c51ea..a0e7c17 100644 --- a/backend/app/api/settings.py +++ b/backend/app/api/settings.py @@ -319,6 +319,52 @@ class MinuteSyncPrefs(BaseModel): minute_sync_days: int = 5 +class DataProvidersIn(BaseModel): + daily_data_provider: str | None = None + adj_factor_provider: str | None = None + minute_data_provider: str | None = None + realtime_data_provider: str | None = None + financial_data_provider: str | None = None + + +class CustomSourceTestIn(BaseModel): + provider: str + dataset: str + symbols: list[str] | None = None + + +class DatasetFieldMapItem(BaseModel): + source: str + target: str + + +class DatasetConfigIn(BaseModel): + url: str + method: str = "GET" + batch: int | None = None + rpm: int | None = None + response_path: str = "" + field_map: dict[str, str] = {} + transforms: dict[str, str] = {} + symbols_param: str = "symbols" + start_param: str = "start_time" + end_param: str = "end_time" + + +class AuthConfigIn(BaseModel): + type: str = "none" + token_env: str | None = None + header: str = "Authorization" + param: str = "token" + + +class CustomSourceIn(BaseModel): + name: str + display_name: str = "" + auth: AuthConfigIn = AuthConfigIn() + datasets: dict[str, DatasetConfigIn] = {} + + @router.get("/preferences") def get_preferences() -> dict: """返回用户偏好设置。""" @@ -333,6 +379,7 @@ def get_preferences() -> dict: "adj_factor_provider": preferences.get_adj_factor_provider(), "minute_data_provider": preferences.get_minute_data_provider(), "realtime_data_provider": preferences.get_realtime_data_provider(), + "financial_data_provider": preferences.get_financial_provider(), "realtime_watchlist_symbols": preferences.get_realtime_watchlist_symbols(), **preferences.get_realtime_quote_scope(), "pipeline_pull_a_share": preferences.get_pipeline_pull_a_share(), @@ -364,6 +411,106 @@ def get_preferences() -> dict: } +@router.get("/data-sources") +def list_data_sources() -> dict: + """列出已加载的自定义数据源。""" + from app.data_providers import custom as custom_sources + return { + "builtin": [{"name": "tickflow", "display_name": "TickFlow", "datasets": ["daily", "adj_factor", "realtime"]}], + "custom": custom_sources.list_sources(), + "errors": custom_sources.errors(), + "config_dir": str(custom_sources.data_sources_dir()), + } + + +@router.post("/data-sources/reload") +def reload_data_sources() -> dict: + """重新加载 data_sources/*.yaml。""" + from app.data_providers import custom as custom_sources + custom_sources.load_all() + return list_data_sources() + + +@router.get("/data-sources/{name}") +def get_data_source(name: str) -> dict: + """读取一个自定义数据源的完整配置(用于前端编辑回填)。""" + from app.data_providers import custom as custom_sources + cfg = custom_sources.get_config_dict(name) + if cfg is None: + raise HTTPException(status_code=404, detail=f"数据源 '{name}' 不存在") + return cfg + + +@router.post("/data-sources") +def save_data_source(req: CustomSourceIn) -> dict: + """创建或更新一个自定义数据源 yaml, 保存后自动 reload。""" + from app.data_providers import custom as custom_sources + config = req.model_dump() + config["name"] = (config.get("name") or "").lower() + try: + custom_sources.save_config(config["name"], config) + custom_sources.load_all() + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) from e + return list_data_sources() + + +@router.delete("/data-sources/{name}") +def delete_data_source(name: str) -> dict: + """删除一个自定义数据源 yaml, 保存后自动 reload。 + + 若当前总开关选中的就是被删的源, 回退到 tickflow。 + """ + from app.data_providers import custom as custom_sources + from app.services import preferences + try: + custom_sources.delete_config(name) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) from e + custom_sources.load_all() + # 回退被删源的偏好 + updates: dict = {} + if preferences.get_daily_data_provider() == name: + updates["daily_data_provider"] = "tickflow" + if preferences.get_realtime_data_provider() == name: + updates["realtime_data_provider"] = "tickflow" + if preferences.get_financial_provider() == name: + updates["financial_data_provider"] = "tickflow" + adj = preferences.get_adj_factor_provider() + if adj == name: + updates["adj_factor_provider"] = "same_as_daily" + if updates: + preferences.save(updates) + return list_data_sources() + + +@router.post("/data-sources/test") +def test_data_source(req: CustomSourceTestIn) -> dict: + """试拉自定义数据源，不写盘。""" + from app.data_providers import custom as custom_sources + provider = custom_sources.get_provider(req.provider) + try: + return provider.test_dataset(req.dataset, req.symbols) + except Exception as e: # noqa: BLE001 + raise HTTPException(status_code=400, detail=f"自定义数据源测试失败: {e}") from e + + +@router.put("/preferences/data-providers") +def update_data_providers(req: DataProvidersIn) -> dict: + """保存数据源选择。""" + from app.services import preferences + updates = req.model_dump(exclude_none=True) + if updates: + preferences.save(updates) + return { + "daily_data_provider": preferences.get_daily_data_provider(), + "adj_factor_provider": preferences.get_adj_factor_provider(), + "minute_data_provider": preferences.get_minute_data_provider(), + "realtime_data_provider": preferences.get_realtime_data_provider(), + "financial_data_provider": preferences.get_financial_provider(), + } + + @router.get("/preferences/watchlist-columns") def get_watchlist_columns() -> dict: """返回自选列表列配置。""" diff --git a/backend/app/data_providers/custom/__init__.py b/backend/app/data_providers/custom/__init__.py new file mode 100644 index 0000000..66870f1 --- /dev/null +++ b/backend/app/data_providers/custom/__init__.py @@ -0,0 +1,28 @@ +"""Custom data source extension points.""" +from app.data_providers.custom.loader import ( + data_sources_dir, + delete_config, + errors, + get_config_dict, + get_provider, + is_custom_provider, + list_sources, + load_all, + names, + provider_has_dataset, + save_config, +) + +__all__ = [ + "data_sources_dir", + "delete_config", + "errors", + "get_config_dict", + "get_provider", + "is_custom_provider", + "list_sources", + "load_all", + "names", + "provider_has_dataset", + "save_config", +] diff --git a/backend/app/data_providers/custom/config.py b/backend/app/data_providers/custom/config.py new file mode 100644 index 0000000..078d650 --- /dev/null +++ b/backend/app/data_providers/custom/config.py @@ -0,0 +1,92 @@ +"""Custom HTTP data source configuration.""" +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Literal + +import yaml + +DatasetName = Literal["daily", "adj_factor", "realtime", "minute", "financial"] + + +@dataclass(frozen=True) +class AuthConfig: + type: str = "none" + token_env: str | None = None + header: str = "Authorization" + param: str = "token" + + +@dataclass(frozen=True) +class DatasetConfig: + url: str + method: str = "GET" + batch: int | None = None + rpm: int | None = None + timeout: float = 30.0 + response_path: str = "" + field_map: dict[str, str] = field(default_factory=dict) + transforms: dict[str, str] = field(default_factory=dict) + params: dict[str, Any] = field(default_factory=dict) + body: dict[str, Any] = field(default_factory=dict) + symbols_param: str = "symbols" + start_param: str = "start_time" + end_param: str = "end_time" + + +@dataclass(frozen=True) +class CustomSourceConfig: + name: str + display_name: str + auth: AuthConfig = field(default_factory=AuthConfig) + datasets: dict[str, DatasetConfig] = field(default_factory=dict) + path: Path | None = None + + def has_dataset(self, name: DatasetName) -> bool: + return name in self.datasets + + +def _auth_from_dict(raw: dict[str, Any] | None) -> AuthConfig: + raw = raw or {} + return AuthConfig( + type=str(raw.get("type", "none") or "none").lower(), + token_env=raw.get("token_env"), + header=str(raw.get("header", "Authorization") or "Authorization"), + param=str(raw.get("param", "token") or "token"), + ) + + +def _dataset_from_dict(raw: dict[str, Any]) -> DatasetConfig: + return DatasetConfig( + url=str(raw.get("url", "") or ""), + method=str(raw.get("method", "GET") or "GET").upper(), + batch=int(raw["batch"]) if raw.get("batch") is not None else None, + rpm=int(raw["rpm"]) if raw.get("rpm") is not None else None, + timeout=float(raw.get("timeout", 30.0) or 30.0), + response_path=str(raw.get("response_path", "") or ""), + field_map={str(k): str(v) for k, v in (raw.get("field_map") or {}).items()}, + transforms={str(k): str(v) for k, v in (raw.get("transforms") or {}).items()}, + params=dict(raw.get("params") or {}), + body=dict(raw.get("body") or {}), + symbols_param=str(raw.get("symbols_param", "symbols") or "symbols"), + start_param=str(raw.get("start_param", "start_time") or "start_time"), + end_param=str(raw.get("end_param", "end_time") or "end_time"), + ) + + +def load_config(path: Path) -> CustomSourceConfig: + raw = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + datasets = { + name: _dataset_from_dict(cfg) + for name, cfg in (raw.get("datasets") or {}).items() + if name in {"daily", "adj_factor", "realtime", "minute", "financial"} and isinstance(cfg, dict) + } + name = str(raw.get("name", path.stem) or path.stem).lower() + return CustomSourceConfig( + name=name, + display_name=str(raw.get("display_name", name) or name), + auth=_auth_from_dict(raw.get("auth")), + datasets=datasets, + path=path, + ) diff --git a/backend/app/data_providers/custom/loader.py b/backend/app/data_providers/custom/loader.py new file mode 100644 index 0000000..2b39d86 --- /dev/null +++ b/backend/app/data_providers/custom/loader.py @@ -0,0 +1,228 @@ +"""Load custom data source definitions from user data files.""" +from __future__ import annotations + +import logging +import re +from pathlib import Path + +import yaml + +from app.config import settings +from app.data_providers.custom.config import CustomSourceConfig, load_config +from app.data_providers.custom.provider import GenericHTTPProvider + +logger = logging.getLogger(__name__) + +_PROVIDERS: dict[str, GenericHTTPProvider] = {} +_LOAD_ERRORS: list[dict] = [] + +_NAME_RE = re.compile(r"^[a-z0-9_]+$") + + +def data_sources_dir() -> Path: + return settings.data_dir / "data_sources" + + +def load_all(path: Path | None = None) -> None: + """Load all custom provider YAML files into process memory.""" + global _PROVIDERS, _LOAD_ERRORS + for provider in _PROVIDERS.values(): + provider.close() + _PROVIDERS = {} + _LOAD_ERRORS = [] + + base = path or data_sources_dir() + base.mkdir(parents=True, exist_ok=True) + for file in sorted([*base.glob("*.yaml"), *base.glob("*.yml")]): + try: + config = load_config(file) + provider = GenericHTTPProvider(config) + errors = provider.validate() + if errors: + _LOAD_ERRORS.append({"path": str(file), "name": config.name, "errors": errors}) + provider.close() + continue + _PROVIDERS[config.name] = provider + except Exception as e: # noqa: BLE001 + logger.warning("custom data source load failed %s: %s", file, e) + _LOAD_ERRORS.append({"path": str(file), "errors": [str(e)]}) + + +def list_sources() -> list[dict]: + return [ + { + "name": provider.name, + "display_name": provider.config.display_name, + "datasets": sorted(provider.config.datasets.keys()), + "path": str(provider.config.path) if provider.config.path else None, + } + for provider in _PROVIDERS.values() + ] + + +def names() -> set[str]: + return set(_PROVIDERS) + + +def errors() -> list[dict]: + return list(_LOAD_ERRORS) + + +def get_provider(name: str) -> GenericHTTPProvider: + provider = _PROVIDERS.get((name or "").lower()) + if provider is None: + raise ValueError(f"Custom data source not found or invalid: {name}") + return provider + + +def is_custom_provider(name: str) -> bool: + return (name or "").lower() in _PROVIDERS + + +def provider_has_dataset(name: str, dataset: str) -> bool: + """判断某个 custom 源是否配置了指定数据集。 + + 用于主流程分流: 总开关选了 custom, 但某个数据集未启用时, 该数据集回退 TickFlow。 + """ + provider = _PROVIDERS.get((name or "").lower()) + if provider is None: + return False + return dataset in provider.config.datasets + + +def get_config_dict(name: str) -> dict | None: + """读取一个已加载 custom 源的原始配置 dict(用于前端编辑回填)。""" + provider = _PROVIDERS.get((name or "").lower()) + if provider is None: + return None + return _config_to_dict(provider.config) + + +def _config_to_dict(config: CustomSourceConfig) -> dict: + auth = config.auth + out: dict = { + "name": config.name, + "display_name": config.display_name, + "auth": { + "type": auth.type, + **({"token_env": auth.token_env} if auth.token_env else {}), + **({"header": auth.header} if auth.type in {"bearer", "header"} and auth.header != "Authorization" else {}), + **({"param": auth.param} if auth.type == "query" and auth.param != "token" else {}), + }, + "datasets": {}, + } + for ds_name, ds in config.datasets.items(): + out["datasets"][ds_name] = { + "url": ds.url, + "method": ds.method, + **({"batch": ds.batch} if ds.batch is not None else {}), + **({"rpm": ds.rpm} if ds.rpm is not None else {}), + "response_path": ds.response_path, + "field_map": dict(ds.field_map), + **({"transforms": dict(ds.transforms)} if ds.transforms else {}), + "symbols_param": ds.symbols_param, + "start_param": ds.start_param, + "end_param": ds.end_param, + } + return out + + +def save_config(name: str, config: dict) -> Path: + """把一份配置 dict 写成 data/data_sources/{name}.yaml, 返回写入路径。""" + if not _NAME_RE.match(name or ""): + raise ValueError(f"invalid data source name: {name!r} (only lowercase a-z 0-9 _ allowed)") + base = data_sources_dir() + base.mkdir(parents=True, exist_ok=True) + path = (base / f"{name}.yaml").resolve() + if not path.is_relative_to(base.resolve()): + raise ValueError("invalid data source name: path escape detected") + cleaned = _sanitize_for_yaml(config) + path.write_text(yaml.safe_dump(cleaned, allow_unicode=True, sort_keys=False), encoding="utf-8") + return path + + +def delete_config(name: str) -> bool: + """删除 data/data_sources/{name}.yaml。返回是否真的删除了。""" + if not _NAME_RE.match(name or ""): + raise ValueError(f"invalid data source name: {name!r}") + base = data_sources_dir().resolve() + path = (base / f"{name}.yaml").resolve() + if not path.is_relative_to(base): + raise ValueError("invalid data source name: path escape detected") + if not path.exists(): + return False + path.unlink() + return True + + +def _sanitize_for_yaml(config: dict) -> dict: + """剔除前端可能塞进来的空值/未启用数据集, 保证写入的 yaml 干净。""" + out: dict = { + "name": str(config.get("name", "")).lower(), + "display_name": str(config.get("display_name") or config.get("name", "")), + } + auth_raw = config.get("auth") or {} + auth_type = str(auth_raw.get("type", "none") or "none").lower() + auth: dict = {"type": auth_type} + if auth_type != "none" and auth_raw.get("token_env"): + auth["token_env"] = str(auth_raw["token_env"]) + if auth_type in {"bearer", "header"} and auth_raw.get("header"): + auth["header"] = str(auth_raw["header"]) + if auth_type == "query" and auth_raw.get("param"): + auth["param"] = str(auth_raw["param"]) + out["auth"] = auth + + datasets_out: dict = {} + for ds_name, ds_cfg in (config.get("datasets") or {}).items(): + if ds_name not in {"daily", "adj_factor", "realtime", "minute", "financial"}: + continue + if not isinstance(ds_cfg, dict): + continue + ds = _sanitize_dataset(ds_cfg) + if ds: + datasets_out[ds_name] = ds + out["datasets"] = datasets_out + return out + + +def _sanitize_dataset(ds_cfg: dict) -> dict: + out: dict = {} + url = str(ds_cfg.get("url", "") or "").strip() + if not url: + return out + out["url"] = url + method = str(ds_cfg.get("method", "GET") or "GET").upper() + out["method"] = method + if ds_cfg.get("batch") is not None: + try: + out["batch"] = int(ds_cfg["batch"]) + except (TypeError, ValueError): + pass + if ds_cfg.get("rpm") is not None: + try: + out["rpm"] = int(ds_cfg["rpm"]) + except (TypeError, ValueError): + pass + out["response_path"] = str(ds_cfg.get("response_path", "") or "") + field_map = { + str(k): str(v) + for k, v in (ds_cfg.get("field_map") or {}).items() + if str(k).strip() and str(v).strip() + } + if field_map: + out["field_map"] = field_map + transforms = { + str(k): str(v) + for k, v in (ds_cfg.get("transforms") or {}).items() + if str(k).strip() and str(v).strip() + } + if transforms: + out["transforms"] = transforms + if ds_cfg.get("symbols_param"): + out["symbols_param"] = str(ds_cfg["symbols_param"]) + if ds_cfg.get("start_param"): + out["start_param"] = str(ds_cfg["start_param"]) + if ds_cfg.get("end_param"): + out["end_param"] = str(ds_cfg["end_param"]) + return out + diff --git a/backend/app/data_providers/custom/mapper.py b/backend/app/data_providers/custom/mapper.py new file mode 100644 index 0000000..88f86aa --- /dev/null +++ b/backend/app/data_providers/custom/mapper.py @@ -0,0 +1,81 @@ +"""Mapping helpers for custom data sources.""" +from __future__ import annotations + +from datetime import datetime +from typing import Any + +import polars as pl + + +def extract_rows(payload: Any, response_path: str = "") -> list[dict]: + """Extract a list of row dicts from a JSON payload using dot-path lookup.""" + data = payload + if response_path: + for part in response_path.split("."): + if not part: + continue + if isinstance(data, dict): + data = data.get(part) + else: + data = None + break + if data is None: + return [] + if isinstance(data, dict): + return [data] + if isinstance(data, list): + return [item for item in data if isinstance(item, dict)] + return [] + + +def map_rows(rows: list[dict], field_map: dict[str, str]) -> pl.DataFrame: + if not rows: + return pl.DataFrame() + df = pl.DataFrame(rows) + rename = {src: dst for src, dst in field_map.items() if src in df.columns and src != dst} + if rename: + df = df.rename(rename) + keep = list(dict.fromkeys(field_map.values())) + keep = [col for col in keep if col in df.columns] + return df.select(keep) if keep else pl.DataFrame() + + +def apply_transforms(df: pl.DataFrame, transforms: dict[str, str]) -> pl.DataFrame: + """Apply a small safe transform set. No eval is used.""" + if df.is_empty() or not transforms: + return df + out = df + for col, expr in transforms.items(): + if col not in out.columns: + continue + text = expr.strip() + if text == "value * 100": + out = out.with_columns((pl.col(col).cast(pl.Float64, strict=False) * 100).alias(col)) + elif text == "value / 100": + out = out.with_columns((pl.col(col).cast(pl.Float64, strict=False) / 100).alias(col)) + elif text == "value / 10000": + out = out.with_columns((pl.col(col).cast(pl.Float64, strict=False) / 10000).alias(col)) + elif text.startswith("parse_date("): + fmt = _extract_format(text) or "%Y-%m-%d" + out = out.with_columns( + pl.col(col).cast(pl.Utf8, strict=False).str.strptime(pl.Date, format=fmt, strict=False).alias(col) + ) + elif text.startswith("parse_datetime("): + fmt = _extract_format(text) or "%Y-%m-%d %H:%M:%S" + out = out.with_columns( + pl.col(col).cast(pl.Utf8, strict=False).str.strptime(pl.Datetime, format=fmt, strict=False).alias(col) + ) + return out + + +def _extract_format(expr: str) -> str | None: + for quote in ("'", '"'): + if quote in expr: + parts = expr.split(quote) + if len(parts) >= 3: + return parts[1] + return None + + +def datetime_payload(value: datetime | None) -> str | None: + return value.isoformat() if value else None diff --git a/backend/app/data_providers/custom/provider.py b/backend/app/data_providers/custom/provider.py new file mode 100644 index 0000000..2266892 --- /dev/null +++ b/backend/app/data_providers/custom/provider.py @@ -0,0 +1,272 @@ +"""Generic HTTP provider for custom market data sources.""" +from __future__ import annotations + +import logging +import os +from datetime import datetime +from pathlib import Path +from typing import Any + +import httpx +import polars as pl + +from app.config import settings +from app.data_providers.custom.config import CustomSourceConfig, DatasetConfig +from app.data_providers.custom.mapper import apply_transforms, datetime_payload, extract_rows, map_rows +from app.data_providers.normalizer import normalize_adj_factors, normalize_daily +from app.tickflow.rate_limits import chunked, sleep_between_batches + +logger = logging.getLogger(__name__) + +_REQUIRED = { + "daily": {"symbol", "date", "open", "high", "low", "close", "volume", "amount"}, + "adj_factor": {"symbol", "trade_date", "ex_factor"}, + "realtime": {"symbol", "last_price", "prev_close", "open", "high", "low", "volume"}, + "minute": {"symbol", "datetime", "open", "high", "low", "close", "volume", "amount"}, + # financial 字段由数据源决定, 只要求能映射出 symbol + "financial": {"symbol"}, +} + + +class GenericHTTPProvider: + """HTTP-backed custom source. It only handles fetching and schema mapping.""" + + def __init__(self, config: CustomSourceConfig) -> None: + self.config = config + self.name = config.name + self._client = httpx.Client(timeout=30.0) + + def close(self) -> None: + self._client.close() + + def validate(self) -> list[str]: + errors: list[str] = [] + for dataset, cfg in self.config.datasets.items(): + if not cfg.url: + errors.append(f"{dataset}: url is required") + required = _REQUIRED.get(dataset) + if required: + mapped = set(cfg.field_map.values()) + missing = sorted(required - mapped) + if missing: + errors.append(f"{dataset}: missing mapped fields: {', '.join(missing)}") + return errors + + def get_daily( + self, + symbols: list[str], + start_time: datetime | None, + end_time: datetime | None, + asset_type: str = "stock", # noqa: ARG002 + on_chunk_done=None, + ) -> pl.DataFrame: + cfg = self._dataset("daily") + frames: list[pl.DataFrame] = [] + chunks = chunked(symbols, cfg.batch) + for i, chunk in enumerate(chunks): + sleep_between_batches(i, cfg.rpm) + rows = self._request_rows(cfg, symbols=chunk, start_time=start_time, end_time=end_time) + df = self._mapped_frame(cfg, rows) + df = normalize_daily(df, source=self.name) + if not df.is_empty(): + frames.append(df) + if on_chunk_done: + on_chunk_done(i + 1, len(chunks)) + return pl.concat(frames, how="diagonal_relaxed") if frames else pl.DataFrame() + + def get_adj_factors( + self, + symbols: list[str], + start_time: datetime | None, + end_time: datetime | None, + asset_type: str = "stock", # noqa: ARG002 + on_chunk_done=None, + ) -> pl.DataFrame: + cfg = self._dataset("adj_factor") + frames: list[pl.DataFrame] = [] + chunks = chunked(symbols, cfg.batch) + for i, chunk in enumerate(chunks): + sleep_between_batches(i, cfg.rpm) + rows = self._request_rows(cfg, symbols=chunk, start_time=start_time, end_time=end_time) + df = self._mapped_frame(cfg, rows) + df = normalize_adj_factors(df, source=self.name) + if not df.is_empty(): + frames.append(df) + if on_chunk_done: + on_chunk_done(i + 1, len(chunks)) + return pl.concat(frames, how="diagonal_relaxed") if frames else pl.DataFrame() + + def get_realtime(self) -> list[dict]: + cfg = self._dataset("realtime") + rows = self._request_rows(cfg) + df = self._mapped_frame(cfg, rows) + if df.is_empty(): + return [] + return df.to_dicts() + + def get_minute( + self, + symbols: list[str], + start_time: datetime | None, + end_time: datetime | None, + asset_type: str = "stock", # noqa: ARG002 + on_chunk_done=None, + ) -> pl.DataFrame: + cfg = self._dataset("minute") + frames: list[pl.DataFrame] = [] + chunks = chunked(symbols, cfg.batch) + for i, chunk in enumerate(chunks): + sleep_between_batches(i, cfg.rpm) + rows = self._request_rows(cfg, symbols=chunk, start_time=start_time, end_time=end_time) + df = self._mapped_frame(cfg, rows) + df = self._normalize_minute(df) + if not df.is_empty(): + frames.append(df) + if on_chunk_done: + on_chunk_done(i + 1, len(chunks)) + return pl.concat(frames, how="diagonal_relaxed") if frames else pl.DataFrame() + + def get_financials( + self, + table: str, + symbols: list[str], + latest_only: bool = True, # noqa: ARG002 + ) -> pl.DataFrame: + """拉取财务数据。table ∈ {metrics, income, balance_sheet, cash_flow}。 + + custom 源用一个 'financial' dataset 配置覆盖 4 张表; 请求时把 table 作为参数传给上游, + 上游根据 table 返回对应数据。字段由数据源决定, 这里只确保有 symbol 列。 + """ + cfg = self._dataset("financial") + frames: list[pl.DataFrame] = [] + chunks = chunked(symbols, cfg.batch) + for i, chunk in enumerate(chunks): + sleep_between_batches(i, cfg.rpm) + # 把 table 注入到请求参数 (上游据此区分 4 张表) + extra_params = {**cfg.params, "table": table} + extra_body = {**cfg.body, "table": table} + rows = self._request_rows( + cfg, symbols=chunk, + override_params=extra_params, override_body=extra_body, + ) + df = self._mapped_frame(cfg, rows) + if not df.is_empty(): + frames.append(df) + if not frames: + return pl.DataFrame() + return pl.concat(frames, how="diagonal_relaxed") + + @staticmethod + def _normalize_minute(df: pl.DataFrame) -> pl.DataFrame: + """把映射后的 df 规范成 minute canonical 列。""" + if df.is_empty(): + return df + if "datetime" in df.columns and df.schema["datetime"] != pl.Datetime("us"): + df = df.with_columns(pl.col("datetime").cast(pl.Datetime("us"), strict=False)) + for col in ("open", "high", "low", "close", "volume", "amount"): + if col in df.columns: + df = df.with_columns(pl.col(col).cast(pl.Float64, strict=False)) + keep = [c for c in ("symbol", "datetime", "open", "high", "low", "close", "volume", "amount") if c in df.columns] + return df.select(keep) if keep else pl.DataFrame() + + def test_dataset(self, dataset: str, symbols: list[str] | None = None) -> dict: + cfg = self._dataset(dataset) + rows = self._request_rows(cfg, symbols=symbols or []) + df = self._mapped_frame(cfg, rows) + return { + "provider": self.name, + "dataset": dataset, + "rows": len(rows), + "columns": df.columns, + "preview": df.head(5).to_dicts() if not df.is_empty() else [], + } + + def _dataset(self, name: str) -> DatasetConfig: + cfg = self.config.datasets.get(name) + if not cfg: + raise ValueError(f"Custom data source '{self.name}' does not configure dataset '{name}'") + return cfg + + def _mapped_frame(self, cfg: DatasetConfig, rows: list[dict]) -> pl.DataFrame: + df = map_rows(rows, cfg.field_map) + return apply_transforms(df, cfg.transforms) + + def _request_rows( + self, + cfg: DatasetConfig, + *, + symbols: list[str] | None = None, + start_time: datetime | None = None, + end_time: datetime | None = None, + override_params: dict[str, Any] | None = None, + override_body: dict[str, Any] | None = None, + ) -> list[dict]: + headers, auth_params = self._auth_parts() + params = dict(cfg.params) + params.update(auth_params) + if override_params: + params.update(override_params) + body = dict(cfg.body) + if override_body: + body.update(override_body) + if symbols: + body[cfg.symbols_param] = symbols + params.setdefault(cfg.symbols_param, ",".join(symbols)) + start_value = datetime_payload(start_time) + end_value = datetime_payload(end_time) + if start_value: + body[cfg.start_param] = start_value + params.setdefault(cfg.start_param, start_value) + if end_value: + body[cfg.end_param] = end_value + params.setdefault(cfg.end_param, end_value) + + method = cfg.method.upper() + request_kwargs: dict[str, Any] = {"headers": headers, "timeout": cfg.timeout} + if method == "GET": + request_kwargs["params"] = params + else: + request_kwargs["params"] = auth_params + request_kwargs["json"] = body + resp = self._client.request(method, cfg.url, **request_kwargs) + resp.raise_for_status() + return extract_rows(resp.json(), cfg.response_path) + + def _auth_parts(self) -> tuple[dict[str, str], dict[str, str]]: + auth = self.config.auth + if auth.type == "none": + return {}, {} + token = _token_from_env(auth.token_env) if auth.token_env else None + if not token: + logger.warning("custom data source %s auth token is not set", self.name) + return {}, {} + if auth.type == "bearer": + return {auth.header: f"Bearer {token}"}, {} + if auth.type == "header": + return {auth.header: token}, {} + if auth.type == "query": + return {}, {auth.param: token} + return {}, {} + + +def _token_from_env(name: str | None) -> str | None: + if not name: + return None + token = os.getenv(name) + if token: + return token + candidates = [settings.data_dir.parent / ".env", Path.cwd() / ".env", Path.cwd().parent / ".env"] + env_path = next((path for path in candidates if path.exists()), None) + if env_path is None: + return None + try: + for line in env_path.read_text(encoding="utf-8").splitlines(): + text = line.strip() + if not text or text.startswith("#") or "=" not in text: + continue + key, value = text.split("=", 1) + if key.strip() == name: + return value.strip().strip('"').strip("'") + except Exception: # noqa: BLE001 + return None + return None diff --git a/backend/app/jobs/daily_pipeline.py b/backend/app/jobs/daily_pipeline.py index a8ad488..d722f84 100644 --- a/backend/app/jobs/daily_pipeline.py +++ b/backend/app/jobs/daily_pipeline.py @@ -119,7 +119,7 @@ def run_now( if not pull_a_share: emit("sync_daily", 45, "已跳过 A 股日K同步(拉取内容未勾选)") logger.info("sync_daily: skipped (pipeline_pull_a_share=False)") - elif today_exists and capset.has(Cap.QUOTE_POOL): + elif today_exists and capset.has(Cap.QUOTE_POOL) and _prefs.get_daily_data_provider() == "tickflow": # 付费档:今天有数据(QuoteService 已落盘)→ 实时行情覆写,确保最新。 # free/none 档无 quote.pool 能力,即便今天已有数据(如从 expert 降级), # 也降级到下方 batch 路径刷新,避免调用无权限的实时行情接口。 @@ -179,7 +179,11 @@ def run_now( # (这两类分支不拉历史日K, 除权不能用日K范围, 只能兜底最近几日) written_adj = 0 affected_symbols: list[str] = [] - if capset.has(Cap.ADJ_FACTOR): + adj_provider = _prefs.get_adj_factor_provider() + if adj_provider == "same_as_daily": + adj_provider = _prefs.get_daily_data_provider() + can_sync_adj = capset.has(Cap.ADJ_FACTOR) or adj_provider != "tickflow" + if can_sync_adj: from datetime import datetime, timedelta adj_end = datetime.now() if daily_range_start is not None: diff --git a/backend/app/main.py b/backend/app/main.py index 541732b..3dea8b7 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -60,6 +60,14 @@ async def lifespan(app: FastAPI): app.state.capabilities = capset logger.info("ready; %d capabilities active", len(capset.all())) + # 自定义数据源配置(可选): 失败只记录错误, 不影响 TickFlow 基准路径。 + try: + from app.data_providers import custom as custom_sources + custom_sources.load_all() + logger.info("custom data sources loaded: %d", len(custom_sources.list_sources())) + except Exception as e: # noqa: BLE001 + logger.warning("custom data sources init failed: %s", e) + # 全局行情服务 qs = QuoteService() app.state.quote_service = qs diff --git a/backend/app/services/depth_service.py b/backend/app/services/depth_service.py index fa16803..dd24faf 100644 --- a/backend/app/services/depth_service.py +++ b/backend/app/services/depth_service.py @@ -31,6 +31,9 @@ from pathlib import Path import polars as pl +from app.tickflow.capabilities import Cap +from app.tickflow.rate_limits import chunked, resolve_limit, sleep_between_batches + logger = logging.getLogger(__name__) @@ -269,17 +272,12 @@ class DepthService: tf = get_client() capset = self._get_capset() - lim = capset.limits(__import__("app.tickflow.capabilities", fromlist=["Cap"]).Cap.DEPTH5_BATCH) - batch_size = (lim.batch if lim and lim.batch else 100) - rpm = (lim.rpm if lim and lim.rpm else 30) - # 批间隔 = 60/rpm(匀速) - inter_batch = 60.0 / rpm if rpm > 0 else 2.0 + limit = resolve_limit(capset, Cap.DEPTH5_BATCH, default_batch=100, default_rpm=30) result: dict = {} - chunks = [symbols[i:i + batch_size] for i in range(0, len(symbols), batch_size)] + chunks = chunked(symbols, limit.batch) for i, chunk in enumerate(chunks): - if i > 0: - time.sleep(inter_batch) + sleep_between_batches(i, limit.rpm, default_interval=2.0) try: # SDK 的 batch 内部已按 batch_size 切, 这里再切一层防单请求过大 data = tf.depth.batch(chunk) @@ -529,7 +527,7 @@ class DepthService: # ================================================================ def _notify_takeover(self, n_stocks: int, user_interval: float, actual_interval: float) -> None: - """系统接管通知: 复用 quote_service 的 _pending_alerts 通道。""" + """系统接管通知: 通过 quote_service 广播到所有 SSE 订阅者。""" if not self._app_state: return qs = getattr(self._app_state, "quote_service", None) @@ -543,9 +541,7 @@ class DepthService: "message": msg, } try: - with qs._lock: - qs._pending_alerts.append(alert) - qs._alert_event.set() + qs.push_alerts([alert]) except Exception as e: # noqa: BLE001 logger.debug("depth 接管通知推送失败: %s", e) diff --git a/backend/app/services/extend_history.py b/backend/app/services/extend_history.py index 1339a52..dc0037a 100644 --- a/backend/app/services/extend_history.py +++ b/backend/app/services/extend_history.py @@ -165,7 +165,12 @@ def run_extend_history( adj_start_str = new_start.strftime("%Y-%m-%d") adj_end_str = today.strftime("%Y-%m-%d") - if capset.has(Cap.ADJ_FACTOR): + from app.services import preferences as _prefs + adj_provider = _prefs.get_adj_factor_provider() + if adj_provider == "same_as_daily": + adj_provider = _prefs.get_daily_data_provider() + can_sync_adj = capset.has(Cap.ADJ_FACTOR) or adj_provider != "tickflow" + if can_sync_adj: emit("extend_history", 48, f"获取除权因子 [{adj_start_str} ~ {adj_end_str}]…") logger.info("extend_history: adj_factor [%s ~ %s]", adj_start_str, adj_end_str) diff --git a/backend/app/services/financial_sync.py b/backend/app/services/financial_sync.py index 39f4102..daa20b5 100644 --- a/backend/app/services/financial_sync.py +++ b/backend/app/services/financial_sync.py @@ -42,6 +42,16 @@ def _get_symbols(data_dir: Path) -> list[str]: return [] +def _financial_is_custom() -> bool: + """当前财务数据源是否走 custom (用于绕过 TickFlow Expert 套餐门槛)。""" + from app.services import preferences + provider = preferences.get_financial_provider() + if provider == "tickflow": + return False + from app.data_providers import custom as custom_sources + return custom_sources.provider_has_dataset(provider, "financial") + + def _sync_table( table: str, symbols: list[str], @@ -50,13 +60,28 @@ def _sync_table( latest_only: bool = True, ) -> int: """同步单张财务表。返回写入的行数。""" - if not capset.has(Cap.FINANCIAL): + is_custom = _financial_is_custom() + if not is_custom and not capset.has(Cap.FINANCIAL): logger.info("sync_%s skipped: no FINANCIAL capability", table) return 0 if not symbols: logger.warning("sync_%s skipped: no symbols", table) return 0 + # 自定义数据源分流 + if is_custom: + from app.services import preferences + from app.data_providers import custom as custom_sources + provider = custom_sources.get_provider(preferences.get_financial_provider()) + df = provider.get_financials(table, symbols, latest_only=latest_only) + if df.is_empty() or "symbol" not in df.columns: + return 0 + out_dir = data_dir / "financials" / table + out_dir.mkdir(parents=True, exist_ok=True) + df.write_parquet(out_dir / "part.parquet") + logger.info("sync_%s done via custom: %d records written", table, len(df)) + return len(df) + from app.tickflow.client import get_client tf = get_client() @@ -347,7 +372,7 @@ class FinancialScheduler: 用 _is_syncing 标志防并发:若已有同步在进行,本次直接跳过, 避免重复请求拖慢服务端 / 触发上游限流。 """ - if not self._capset or not self._capset.has(Cap.FINANCIAL): + if not self._capset or (not self._capset.has(Cap.FINANCIAL) and not _financial_is_custom()): return {} with self._lock: if self._is_syncing: @@ -372,7 +397,7 @@ class FinancialScheduler: /status 已能看到 syncing=True,无竞态窗口;同时防止快速重复点击 启动多个后台线程。后台线程复用 _run_body 执行真正的同步逻辑。 """ - if not self._capset or not self._capset.has(Cap.FINANCIAL): + if not self._capset or (not self._capset.has(Cap.FINANCIAL) and not _financial_is_custom()): return {"started": False, "reason": "no FINANCIAL capability"} with self._lock: if self._is_syncing: diff --git a/backend/app/services/index_sync.py b/backend/app/services/index_sync.py index 1f474e3..098ce67 100644 --- a/backend/app/services/index_sync.py +++ b/backend/app/services/index_sync.py @@ -17,6 +17,7 @@ from app.indicators.pipeline import compute_enriched from app.services import kline_sync, preferences from app.tickflow.capabilities import Cap, CapabilitySet from app.tickflow.client import get_client +from app.tickflow.rate_limits import chunked, min_batch, resolve_limit, sleep_between_batches from app.tickflow.repository import KlineRepository logger = logging.getLogger(__name__) @@ -220,22 +221,16 @@ def sync_and_persist_index_daily( if instruments.is_empty() or "symbol" not in instruments.columns: return 0 symbols = sorted(set(instruments["symbol"].to_list())) - lim = capset.limits(Cap.KLINE_DAILY_BATCH) - batch_size = preferences.get_index_daily_batch_size() - if lim and lim.batch: - batch_size = min(batch_size, lim.batch) - rpm = lim.rpm if lim else None + limit = resolve_limit(capset, Cap.KLINE_DAILY_BATCH) + batch_size = min_batch(preferences.get_index_daily_batch_size(), limit) end_time = end_date or datetime.now() start_time = start_date or (end_time - timedelta(days=365)) total_rows = 0 - interval = (60.0 / rpm) if rpm else 0 - chunks = [symbols[i:i + batch_size] for i in range(0, len(symbols), batch_size)] + chunks = chunked(symbols, batch_size) for i, chunk in enumerate(chunks): - if i > 0 and interval > 0: - import time - time.sleep(interval) + sleep_between_batches(i, limit.rpm) raw = kline_sync.sync_daily_batch( chunk, count=count, @@ -318,23 +313,17 @@ def sync_and_persist_etf_daily( if not symbols: return 0 - lim = capset.limits(Cap.KLINE_DAILY_BATCH) - batch_size = preferences.get_index_daily_batch_size() - if lim and lim.batch: - batch_size = min(batch_size, lim.batch) - rpm = lim.rpm if lim else None + limit = resolve_limit(capset, Cap.KLINE_DAILY_BATCH) + batch_size = min_batch(preferences.get_index_daily_batch_size(), limit) end_time = end_date or datetime.now() start_time = start_date or (end_time - timedelta(days=365)) total_rows = 0 - interval = (60.0 / rpm) if rpm else 0 - chunks = [symbols[i:i + batch_size] for i in range(0, len(symbols), batch_size)] + chunks = chunked(symbols, batch_size) factors = _load_etf_factors(repo) for i, chunk in enumerate(chunks): - if i > 0 and interval > 0: - import time - time.sleep(interval) + sleep_between_batches(i, limit.rpm) raw = kline_sync.sync_daily_batch( chunk, count=count, diff --git a/backend/app/services/kline_sync.py b/backend/app/services/kline_sync.py index ba9a751..474142f 100644 --- a/backend/app/services/kline_sync.py +++ b/backend/app/services/kline_sync.py @@ -8,15 +8,16 @@ from __future__ import annotations import logging -import time from collections.abc import Callable from datetime import datetime, timedelta import polars as pl from app.indicators.pipeline import filter_halt_days +from app.services import preferences from app.tickflow.capabilities import Cap, CapabilitySet from app.tickflow.client import get_client +from app.tickflow.rate_limits import chunked, resolve_limit, sleep_between_batches from app.tickflow.repository import KlineRepository logger = logging.getLogger(__name__) @@ -84,16 +85,10 @@ def sync_daily_batch(symbols: list[str], """ tf = get_client() out: list[pl.DataFrame] = [] - interval = (60.0 / rpm) if rpm else 0 - - if batch_size is None: - chunks = [symbols] - else: - chunks = [symbols[i:i + batch_size] for i in range(0, len(symbols), batch_size)] + chunks = chunked(symbols, batch_size) for i, chunk in enumerate(chunks): - if i > 0 and interval > 0: - time.sleep(interval) + sleep_between_batches(i, rpm) try: if start_time and end_time: raw = tf.klines.batch( @@ -141,18 +136,47 @@ def sync_and_persist_daily_batch( start_date/end_date: 外部传入的时间范围(由 pipeline 根据已有数据计算)。 未传入时默认拉最近 1 年。 """ - if not symbols or not capset.has(Cap.KLINE_DAILY_BATCH): + if not symbols: return 0 - lim = capset.limits(Cap.KLINE_DAILY_BATCH) - batch_size = lim.batch if lim and lim.batch else 100 - rpm = lim.rpm if lim else None + provider_name = preferences.get_daily_data_provider() + if provider_name != "tickflow": + from app.data_providers import custom as custom_sources + if custom_sources.provider_has_dataset(provider_name, "daily"): + provider = custom_sources.get_provider(provider_name) + end_time = end_date or datetime.now() + days = count or 365 + start_time = start_date or (end_time - timedelta(days=days)) + df = provider.get_daily( + symbols, + start_time=start_time, + end_time=end_time, + on_chunk_done=on_chunk_done, + ) + if df.is_empty(): + return 0 + repo.append_daily(df) + try: + d = repo.store.data_dir.as_posix() + repo.db.execute( + f"""CREATE OR REPLACE VIEW kline_daily AS + SELECT * FROM read_parquet('{d}/kline_daily/**/*.parquet', union_by_name=true)""" + ) + except Exception as e: # noqa: BLE001 + logger.warning("refresh view failed: %s", e) + return df.height + # 自定义源未配置 daily → 回退 TickFlow + + if not capset.has(Cap.KLINE_DAILY_BATCH): + return 0 + + limit = resolve_limit(capset, Cap.KLINE_DAILY_BATCH, default_batch=100) end_time = end_date or datetime.now() start_time = start_date or (end_time - timedelta(days=365)) df = sync_daily_batch( - symbols, count=count, batch_size=batch_size, rpm=rpm, + symbols, count=count, batch_size=limit.batch, rpm=limit.rpm, start_time=start_time, end_time=end_time, on_chunk_done=on_chunk_done, ) @@ -275,28 +299,65 @@ def sync_adj_factor(symbols: list[str], repo: KlineRepository, 支持增量: 传 start_time/end_time 只拉取该时间范围内的新除权事件。 返回 (写入行数, 受影响的 symbol 列表) — 供 enriched 局部重算使用。 """ - if not capset.has(Cap.ADJ_FACTOR) or not symbols: + if not symbols: + return 0, [] + + provider_name = preferences.get_adj_factor_provider() + if provider_name == "same_as_daily": + provider_name = preferences.get_daily_data_provider() + if provider_name != "tickflow": + from app.data_providers import custom as custom_sources + if custom_sources.provider_has_dataset(provider_name, "adj_factor"): + provider = custom_sources.get_provider(provider_name) + new_data = provider.get_adj_factors( + symbols, + start_time=start_time, + end_time=end_time, + asset_type=asset_type, + on_chunk_done=on_chunk_done, + ) + if new_data.is_empty(): + return 0, [] + affected = new_data["symbol"].unique().to_list() + factor_dir = "adj_factor_etf" if asset_type == "etf" else "adj_factor" + out = repo.store.data_dir / factor_dir / "all.parquet" + out.parent.mkdir(parents=True, exist_ok=True) + if out.exists(): + existing = pl.read_parquet(out) + before = existing.height + merged = pl.concat([existing, new_data]).unique( + subset=["symbol", "trade_date"], keep="last", + ).sort(["symbol", "trade_date"]) + merged.write_parquet(out) + return merged.height - before, affected + new_data.sort(["symbol", "trade_date"]).write_parquet(out) + return new_data.height, affected + # 自定义源未配置 adj_factor → 回退 TickFlow + + if not capset.has(Cap.ADJ_FACTOR): return 0, [] tf = get_client() - lim = capset.limits(Cap.ADJ_FACTOR) - batch_size = lim.batch if lim and lim.batch else 50 - rpm = lim.rpm if lim else 30 - interval = 60.0 / rpm if rpm else 0 + limit = resolve_limit( + capset, + Cap.ADJ_FACTOR, + default_batch=50, + default_rpm=30, + default_rpm_when_unset=False, + ) # 构建 SDK 参数 - sdk_kwargs: dict = {"as_dataframe": True, "batch_size": batch_size, "show_progress": False} + sdk_kwargs: dict = {"as_dataframe": True, "batch_size": limit.batch, "show_progress": False} if start_time: sdk_kwargs["start_time"] = _datetime_to_ms(start_time) if end_time: sdk_kwargs["end_time"] = _datetime_to_ms(end_time) - chunks = [symbols[i:i + batch_size] for i in range(0, len(symbols), batch_size)] + chunks = chunked(symbols, limit.batch) all_dfs: list[pl.DataFrame] = [] for i, chunk in enumerate(chunks): - if i > 0 and interval > 0: - time.sleep(interval) + sleep_between_batches(i, limit.rpm) try: raw = tf.klines.ex_factors(chunk, **sdk_kwargs) normalized = _normalize_adj_factor(raw) @@ -417,18 +478,23 @@ def sync_minute_batch( count 仅作为 fallback 保留。 on_chunk_done(current, total) 每个 chunk 完成后回调。 """ + # 自定义数据源分流: minute provider + provider_name = preferences.get_minute_data_provider() + if provider_name != "tickflow": + from app.data_providers import custom as custom_sources + if custom_sources.provider_has_dataset(provider_name, "minute"): + provider = custom_sources.get_provider(provider_name) + return provider.get_minute( + symbols, start_time=start_time, end_time=end_time, on_chunk_done=on_chunk_done, + ) + # 未配置 minute → 回退 TickFlow + tf = get_client() out: list[pl.DataFrame] = [] - interval = (60.0 / rpm) if rpm else 0 - - if batch_size is None: - chunks = [symbols] - else: - chunks = [symbols[i:i + batch_size] for i in range(0, len(symbols), batch_size)] + chunks = chunked(symbols, batch_size) for i, chunk in enumerate(chunks): - if i > 0 and interval > 0: - time.sleep(interval) + sleep_between_batches(i, rpm) try: if start_time and end_time: raw = tf.klines.batch( @@ -608,7 +674,14 @@ def sync_and_persist_minute( 使用 start_time / end_time 区间拉取, 确保所有标的覆盖同一时间段。 on_chunk_done(current, total) 每个 chunk 完成后回调。 """ - if not symbols or not capset.has(Cap.KLINE_MINUTE_BATCH): + minute_provider = preferences.get_minute_data_provider() + minute_is_custom = False + if minute_provider != "tickflow": + from app.data_providers import custom as custom_sources + minute_is_custom = custom_sources.provider_has_dataset(minute_provider, "minute") + if not symbols: + return 0 + if not minute_is_custom and not capset.has(Cap.KLINE_MINUTE_BATCH): return 0 # 迁移:旧版 _normalize_minute 未转换 timestamp→datetime,导致全部 datetime 为 null @@ -628,12 +701,16 @@ def sync_and_persist_minute( start_time = now - timedelta(days=days) end_time = now - lim = capset.limits(Cap.KLINE_MINUTE_BATCH) - batch_size = lim.batch if lim and lim.batch else 100 - rpm = lim.rpm if lim else 30 + limit = resolve_limit( + capset, + Cap.KLINE_MINUTE_BATCH, + default_batch=100, + default_rpm=30, + default_rpm_when_unset=False, + ) df = sync_minute_batch(symbols, start_time=start_time, end_time=end_time, - batch_size=batch_size, rpm=rpm, + batch_size=limit.batch, rpm=limit.rpm, on_chunk_done=on_chunk_done) if df.is_empty(): return 0 diff --git a/backend/app/services/preferences.py b/backend/app/services/preferences.py index f4e31db..8efe686 100644 --- a/backend/app/services/preferences.py +++ b/backend/app/services/preferences.py @@ -99,26 +99,39 @@ def get_minute_sync_days() -> int: _ALLOWED_DATA_PROVIDERS = {"tickflow"} +def _allowed_data_providers() -> set[str]: + try: + from app.data_providers import custom as custom_sources + return _ALLOWED_DATA_PROVIDERS | custom_sources.names() + except Exception: # noqa: BLE001 + return set(_ALLOWED_DATA_PROVIDERS) + + def get_daily_data_provider() -> str: provider = str(load().get("daily_data_provider", "tickflow") or "tickflow").lower() - return provider if provider in _ALLOWED_DATA_PROVIDERS else "tickflow" + return provider if provider in _allowed_data_providers() else "tickflow" def get_adj_factor_provider() -> str: provider = str(load().get("adj_factor_provider", "same_as_daily") or "same_as_daily").lower() if provider == "same_as_daily": return provider - return provider if provider in _ALLOWED_DATA_PROVIDERS else "same_as_daily" + return provider if provider in _allowed_data_providers() else "same_as_daily" def get_minute_data_provider() -> str: provider = str(load().get("minute_data_provider", "tickflow") or "tickflow").lower() - return provider if provider in _ALLOWED_DATA_PROVIDERS else "tickflow" + return provider if provider in _allowed_data_providers() else "tickflow" def get_realtime_data_provider() -> str: - # 盘中实时现阶段仅支持 TickFlow。 - return "tickflow" + provider = str(load().get("realtime_data_provider", "tickflow") or "tickflow").lower() + return provider if provider in _allowed_data_providers() else "tickflow" + + +def get_financial_provider() -> str: + provider = str(load().get("financial_data_provider", "tickflow") or "tickflow").lower() + return provider if provider in _allowed_data_providers() else "tickflow" # ===== 盘后管道拉取内容开关 (A股 / ETF / 指数 独立控制) ===== diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py index 2b96bb4..da32c6c 100644 --- a/backend/app/services/quote_service.py +++ b/backend/app/services/quote_service.py @@ -90,6 +90,12 @@ class QuoteSubscriber: self._reviews = self._reviews[-self._max_reviews:] self._event.set() + def clear_alerts(self) -> None: + with self._lock: + self._alerts = [] + if not self._quote_updated and not self._depth_updated and not self._reviews: + self._event.clear() + def notify_quote(self) -> None: with self._lock: self._quote_updated = True @@ -266,6 +272,13 @@ class QuoteService: for sub in self._snapshot_subscribers(): sub.push_alerts(alerts) + def push_alerts(self, alerts: list[dict]) -> None: + self._broadcast_alerts(alerts) + + def clear_pending_alerts(self) -> None: + for sub in self._snapshot_subscribers(): + sub.clear_alerts() + def push_review_event(self, event_json: str) -> None: """广播一条复盘进度事件(JSON 字符串), 唤醒所有 SSE generator。 @@ -288,6 +301,9 @@ class QuoteService: @classmethod def realtime_mode(cls) -> str: """当前实时行情模式: none / watchlist / full_market。""" + from app.services import preferences + if preferences.get_realtime_data_provider() != "tickflow": + return "full_market" tier = cls._current_tier() if tier == "none": return "none" @@ -407,6 +423,23 @@ class QuoteService: def _fetch_full_market_quotes(self) -> None: """拉取全市场行情 → 写 daily + 计算 enriched + 更新缓存。""" + from app.services import preferences + + provider_name = preferences.get_realtime_data_provider() + if provider_name != "tickflow": + from app.data_providers import custom as custom_sources + if custom_sources.provider_has_dataset(provider_name, "realtime"): + try: + t0 = time.perf_counter() + now_ts = time.perf_counter() + records = custom_sources.get_provider(provider_name).get_realtime() + except Exception as e: # noqa: BLE001 + logger.warning("自定义实时行情拉取失败: %s", e) + return + self._process_full_market_records(records, t0=t0, now_ts=now_ts) + return + # 自定义源未配置 realtime → 回退 TickFlow + from app.tickflow.client import get_paid_realtime_client tf = get_paid_realtime_client() @@ -480,6 +513,24 @@ class QuoteService: "session": q.get("session"), }) + self._process_full_market_records(records, t0=t0, now_ts=now_ts) + + def _process_full_market_records(self, records: list[dict], *, t0: float, now_ts: float) -> None: + """把全市场 records 写盘并增量计算 enriched。""" + from app.services import preferences + all_index_symbols = set(self._repo.get_index_symbol_set()) if self._repo else set() + core_index_symbols = set(preferences.get_realtime_index_symbols() or self.CORE_INDEX_SYMBOLS) + all_index_symbols.update(core_index_symbols) + all_etf_symbols = set() + if self._repo: + etf_inst = self._repo.get_etf_instruments() + if not etf_inst.is_empty() and "symbol" in etf_inst.columns: + all_etf_symbols = set(etf_inst["symbol"].cast(pl.Utf8).to_list()) + + if not records: + logger.warning("行情数据为空") + return + index_records = [r for r in records if r.get("symbol") in all_index_symbols] etf_records = [r for r in records if r.get("symbol") in all_etf_symbols] stock_records = [ diff --git a/backend/app/services/watchlist.py b/backend/app/services/watchlist.py index 7515b72..476a56c 100644 --- a/backend/app/services/watchlist.py +++ b/backend/app/services/watchlist.py @@ -13,6 +13,7 @@ import polars as pl from app.config import settings from app.tickflow.capabilities import Cap, CapabilitySet from app.tickflow.client import get_client +from app.tickflow.rate_limits import chunked, resolve_limit logger = logging.getLogger(__name__) @@ -104,19 +105,16 @@ def fetch_quotes(symbols: list[str], capset: CapabilitySet, timeout_s: float = 8 quotes: list[dict] = [] # 走 batch - batch_size = 5 if capset.has(Cap.QUOTE_BATCH): - lim = capset.limits(Cap.QUOTE_BATCH) - batch_size = lim.batch if lim and lim.batch else 50 + batch_size = resolve_limit(capset, Cap.QUOTE_BATCH, default_batch=50).batch elif capset.has(Cap.QUOTE_BY_SYMBOL): - lim = capset.limits(Cap.QUOTE_BY_SYMBOL) - batch_size = lim.batch if lim and lim.batch else 5 + batch_size = resolve_limit(capset, Cap.QUOTE_BY_SYMBOL, default_batch=5).batch else: # 无任何实时行情能力(none/free 档走 free-api 服务器,不提供实时行情) # 提前返回空,避免发起注定失败的请求 return [] - chunks = [symbols[i:i + batch_size] for i in range(0, len(symbols), batch_size)] + chunks = chunked(symbols, batch_size) # 用线程池为每个批次加超时保护 pool = ThreadPoolExecutor(max_workers=1) diff --git a/backend/app/tickflow/rate_limits.py b/backend/app/tickflow/rate_limits.py new file mode 100644 index 0000000..86c13e3 --- /dev/null +++ b/backend/app/tickflow/rate_limits.py @@ -0,0 +1,64 @@ +"""TickFlow capability rate-limit helpers. + +This module centralizes the small pieces of batch/rpm resolution used by +TickFlow-backed services. It intentionally does not manage custom data sources. +""" +from __future__ import annotations + +import time +from dataclasses import dataclass +from typing import TypeVar + +from app.tickflow.capabilities import Cap, CapabilitySet + +T = TypeVar("T") + + +@dataclass(frozen=True) +class ResolvedLimit: + batch: int | None + rpm: int | None + + +def resolve_limit( + capset: CapabilitySet, + cap: Cap, + *, + default_batch: int | None = None, + default_rpm: int | None = None, + default_rpm_when_unset: bool = True, +) -> ResolvedLimit: + """Return a capability's batch/rpm with caller-provided fallbacks.""" + lim = capset.limits(cap) + if lim is None: + return ResolvedLimit(batch=default_batch, rpm=default_rpm) + return ResolvedLimit( + batch=lim.batch if lim.batch else default_batch, + rpm=lim.rpm if lim.rpm else (default_rpm if default_rpm_when_unset else None), + ) + + +def batch_interval(rpm: int | None, *, default: float = 0.0) -> float: + """Return the existing uniform batch interval formula: 60 / rpm.""" + return 60.0 / rpm if rpm and rpm > 0 else default + + +def chunked(items: list[T], batch_size: int | None) -> list[list[T]]: + """Split items by batch size, preserving the existing None-as-one-batch behavior.""" + if batch_size is None: + return [items] + return [items[i:i + batch_size] for i in range(0, len(items), batch_size)] + + +def sleep_between_batches(index: int, rpm: int | None, *, default_interval: float = 0.0) -> None: + """Sleep before every batch after the first, using the existing interval formula.""" + if index <= 0: + return + interval = batch_interval(rpm, default=default_interval) + if interval > 0: + time.sleep(interval) + + +def min_batch(preferred: int, limit: ResolvedLimit) -> int: + """Clamp a user-preferred batch size by a resolved capability batch limit.""" + return min(preferred, limit.batch) if limit.batch else preferred diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 62e3677..2267543 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "tickflow-stock-panel-backend" -version = "0.1.70" +version = "0.1.80" description = "A 股选股 + 监控 + 回测面板 — TickFlow 适配" readme = "../README.md" requires-python = ">=3.11" diff --git a/backend/uv.lock b/backend/uv.lock index 5160cb6..2f3bc0b 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -2491,7 +2491,7 @@ all = [ [[package]] name = "tickflow-stock-panel-backend" -version = "0.1.67" +version = "0.1.80" source = { editable = "." } dependencies = [ { name = "apscheduler" }, diff --git a/docs/custom-data-source.md b/docs/custom-data-source.md new file mode 100644 index 0000000..a6c41f8 --- /dev/null +++ b/docs/custom-data-source.md @@ -0,0 +1,205 @@ +# 自定义数据源接入 + +本项目默认使用 TickFlow。自定义数据源是一个可选扩展: 外部 HTTP 服务负责取数和整理, 本项目只把返回结果映射成内部标准字段, 然后复用现有存储、指标、enriched、策略和前端展示逻辑。 + +## 支持范围 + +当前自定义源支持三类数据: + +| 数据集 | 配置名 | 说明 | +| --- | --- | --- | +| 日K | `daily` | 批量返回一组股票在指定区间内的日K | +| 除权因子 | `adj_factor` | 批量返回一组股票的复权因子 | +| 实时行情 | `realtime` | 返回全市场快照,用于盘中 enriched 增量计算 | + +分钟K、财务、深度盘口暂时仍走 TickFlow。 + +## 配置位置 + +把 YAML 放到运行数据目录下: + +```text +data/data_sources/*.yaml +``` + +在桌面版中,`data/` 位于程序目录旁;在开发环境中,通常是项目根目录的 `data/`。 + +修改 YAML 后可在「设置 -> 数据源」点击「重新加载」,或调用: + +```bash +curl -X POST http://127.0.0.1:3018/api/settings/data-sources/reload +``` + +## 最小 YAML + +```yaml +name: mock_source +display_name: "Mock 自定义数据源" +auth: + type: none + +datasets: + daily: + url: http://127.0.0.1:3021/daily + method: POST + batch: 100 + rpm: 200 + response_path: data + field_map: + ts_code: symbol + trade_date: date + open: open + high: high + low: low + close: close + vol: volume + amt: amount + transforms: + date: "parse_date(value, '%Y-%m-%d')" + + adj_factor: + url: http://127.0.0.1:3021/adj_factor + method: POST + batch: 100 + rpm: 200 + response_path: data + field_map: + ts_code: symbol + trade_date: trade_date + factor: ex_factor + transforms: + trade_date: "parse_date(value, '%Y-%m-%d')" + + realtime: + url: http://127.0.0.1:3021/realtime + method: GET + rpm: 60 + response_path: data + field_map: + ts_code: symbol + name: name + last: last_price + pre_close: prev_close + open: open + high: high + low: low + vol: volume + amt: amount + pct: change_pct + amount_change: change_amount + amplitude: amplitude + turnover: turnover_rate +``` + +## 字段契约 + +### daily 必填 + +| 内部字段 | 含义 | +| --- | --- | +| `symbol` | 标准代码,如 `000001.SZ` | +| `date` | 交易日 | +| `open` / `high` / `low` / `close` | 不复权 OHLC | +| `volume` | 成交量 | +| `amount` | 成交额 | + +### adj_factor 必填 + +| 内部字段 | 含义 | +| --- | --- | +| `symbol` | 标准代码 | +| `trade_date` | 除权日期 | +| `ex_factor` | 复权因子 | + +### realtime 必填 + +| 内部字段 | 含义 | +| --- | --- | +| `symbol` | 标准代码 | +| `last_price` | 最新价 | +| `prev_close` | 昨收 | +| `open` / `high` / `low` | 当日 OHLC | +| `volume` | 成交量 | + +建议实时接口额外提供 `amount`、`change_pct`、`change_amount`、`amplitude`、`turnover_rate`、`name`。缺失时部分字段会由 pipeline 回算,但精度取决于可用输入。 + +`change_pct` 和 `amplitude` 使用小数制,例如 `0.0366` 表示 `3.66%`。 + +## 请求约定 + +- `daily` / `adj_factor` 会按 `batch` 切分 symbols。 +- POST 请求会发送 JSON body: `symbols`、`start_time`、`end_time`。 +- GET 请求会发送 query 参数: `symbols=000001.SZ,600000.SH`。 +- `realtime` 必须是全市场快照接口,不支持逐个 symbol 拉实时行情。 + +可通过这些字段改参数名: + +```yaml +symbols_param: symbols +start_param: start_time +end_param: end_time +``` + +## 鉴权 + +支持三种简单鉴权: + +```yaml +auth: + type: bearer + token_env: MY_DATA_TOKEN +``` + +```yaml +auth: + type: header + header: X-Token + token_env: MY_DATA_TOKEN +``` + +```yaml +auth: + type: query + param: token + token_env: MY_DATA_TOKEN +``` + +Token 可以放在系统环境变量或项目 `.env` 中。 + +## 联调流程 + +1. 启动 mock 数据源: + +```bash +cd docs/examples/custom-data-source +python mock_server.py +``` + +2. 复制示例配置: + +```bash +mkdir -p data/data_sources +cp docs/examples/custom-data-source/mock_source.yaml data/data_sources/mock_source.yaml +``` + +3. 在「设置 -> 数据源」点击「重新加载」。 + +4. 使用「试拉测试」选择 `mock_source` 和 `daily` / `adj_factor` / `realtime`。 + +5. 保存数据源选择: + +- 日K: `mock_source` +- 除权因子: `same_as_daily` 或 `mock_source` +- 实时行情: `mock_source` + +6. 触发同步或开启实时行情。 + +## 常见错误 + +| 现象 | 处理 | +| --- | --- | +| 列表里没有 custom 源 | 检查 YAML 是否放在 `data/data_sources/` 并点击重新加载 | +| errors 提示 missing mapped fields | `field_map` 没映射到必填内部字段 | +| 试拉 rows 为 0 | 检查 `response_path` 是否指向数组 | +| 日期列全为空 | 检查 `parse_date` 的格式是否和返回值一致 | +| 实时行情没刷新 | 确认实时数据源已保存为 custom,且返回全市场快照 | diff --git a/docs/examples/custom-data-source/README.md b/docs/examples/custom-data-source/README.md new file mode 100644 index 0000000..215a4f4 --- /dev/null +++ b/docs/examples/custom-data-source/README.md @@ -0,0 +1,43 @@ +# 自定义数据源 mock 联调示例 + +这个目录提供一个本地 mock HTTP 数据源,用于验证项目的自定义数据源接入链路。 + +## 运行 mock 服务 + +```bash +cd docs/examples/custom-data-source +python mock_server.py +``` + +服务默认监听: + +```text +http://127.0.0.1:3021 +``` + +端点: + +| 端点 | 数据 | +| --- | --- | +| `/daily` | 日K | +| `/adj_factor` | 除权因子 | +| `/realtime` | 全市场实时快照 | + +## 接入项目 + +复制示例 YAML 到运行数据目录: + +```bash +mkdir -p data/data_sources +cp docs/examples/custom-data-source/mock_source.yaml data/data_sources/mock_source.yaml +``` + +然后在项目里打开: + +```text +设置 -> 数据源 -> 重新加载 +``` + +选择 `mock_source` 后,可用「试拉测试」验证 `daily`、`adj_factor` 和 `realtime`。 + +完整说明见 [../../custom-data-source.md](../../custom-data-source.md)。 diff --git a/docs/examples/custom-data-source/mock_server.py b/docs/examples/custom-data-source/mock_server.py new file mode 100644 index 0000000..ae31d99 --- /dev/null +++ b/docs/examples/custom-data-source/mock_server.py @@ -0,0 +1,129 @@ +"""Mock custom market data source for local integration tests. + +Run: + python mock_server.py + +Then copy mock_source.yaml to data/data_sources/mock_source.yaml and reload data +sources in the app settings page. +""" +from __future__ import annotations + +from datetime import date, datetime, timedelta +from typing import Any + +import uvicorn +from fastapi import FastAPI, Request + +app = FastAPI(title="Mock Custom Market Data Source") + +SYMBOLS = { + "000001.SZ": "平安银行", + "600000.SH": "浦发银行", + "300750.SZ": "宁德时代", +} + +BASE = { + "000001.SZ": 10.20, + "600000.SH": 8.60, + "300750.SZ": 186.00, +} + + +def _parse_symbols(value: Any) -> list[str]: + if isinstance(value, list): + return [str(v) for v in value if str(v) in SYMBOLS] + if isinstance(value, str) and value: + return [s.strip() for s in value.split(",") if s.strip() in SYMBOLS] + return list(SYMBOLS) + + +async def _payload(request: Request) -> dict: + if request.method == "POST": + try: + return await request.json() + except Exception: + return {} + return dict(request.query_params) + + +def _parse_date(value: Any, fallback: date) -> date: + if not value: + return fallback + text = str(value)[:10] + try: + return date.fromisoformat(text) + except ValueError: + return fallback + + +@app.api_route("/daily", methods=["GET", "POST"]) +async def daily(request: Request): + body = await _payload(request) + symbols = _parse_symbols(body.get("symbols")) + end = _parse_date(body.get("end_time"), date.today()) + start = _parse_date(body.get("start_time"), end - timedelta(days=5)) + rows = [] + cur = start + while cur <= end: + if cur.weekday() < 5: + offset = (cur - start).days + for sym in symbols: + base = BASE[sym] + offset * 0.03 + rows.append({ + "ts_code": sym, + "trade_date": cur.isoformat(), + "open": round(base, 2), + "high": round(base * 1.015, 2), + "low": round(base * 0.985, 2), + "close": round(base * 1.004, 2), + "vol": 120000 + offset * 1000, + "amt": round((120000 + offset * 1000) * base, 2), + }) + cur += timedelta(days=1) + return {"code": 0, "data": rows} + + +@app.api_route("/adj_factor", methods=["GET", "POST"]) +async def adj_factor(request: Request): + body = await _payload(request) + symbols = _parse_symbols(body.get("symbols")) + today = date.today() + return { + "code": 0, + "data": [ + {"ts_code": sym, "trade_date": today.isoformat(), "factor": 1.0} + for sym in symbols + ], + } + + +@app.api_route("/realtime", methods=["GET", "POST"]) +async def realtime(): + now = datetime.now().isoformat(timespec="seconds") + rows = [] + for i, (sym, name) in enumerate(SYMBOLS.items()): + prev = BASE[sym] + last = round(prev * (1 + (i + 1) * 0.006), 2) + change = round(last - prev, 2) + rows.append({ + "ts_code": sym, + "name": name, + "last": last, + "pre_close": prev, + "open": round(prev * 1.002, 2), + "high": round(last * 1.01, 2), + "low": round(prev * 0.99, 2), + "vol": 150000 + i * 20000, + "amt": round((150000 + i * 20000) * last, 2), + "pct": change / prev, + "amount_change": change, + "amplitude": 0.025, + "turnover": 0.012 + i * 0.001, + "timestamp": now, + "session": "regular", + }) + return {"code": 0, "data": rows} + + +if __name__ == "__main__": + uvicorn.run(app, host="127.0.0.1", port=3021) diff --git a/docs/examples/custom-data-source/mock_source.yaml b/docs/examples/custom-data-source/mock_source.yaml new file mode 100644 index 0000000..e277015 --- /dev/null +++ b/docs/examples/custom-data-source/mock_source.yaml @@ -0,0 +1,58 @@ +name: mock_source +display_name: "Mock 自定义数据源" +auth: + type: none + +datasets: + daily: + url: http://127.0.0.1:3021/daily + method: POST + batch: 100 + rpm: 200 + response_path: data + field_map: + ts_code: symbol + trade_date: date + open: open + high: high + low: low + close: close + vol: volume + amt: amount + transforms: + date: "parse_date(value, '%Y-%m-%d')" + + adj_factor: + url: http://127.0.0.1:3021/adj_factor + method: POST + batch: 100 + rpm: 200 + response_path: data + field_map: + ts_code: symbol + trade_date: trade_date + factor: ex_factor + transforms: + trade_date: "parse_date(value, '%Y-%m-%d')" + + realtime: + url: http://127.0.0.1:3021/realtime + method: GET + rpm: 60 + response_path: data + field_map: + ts_code: symbol + name: name + last: last_price + pre_close: prev_close + open: open + high: high + low: low + vol: volume + amt: amount + pct: change_pct + amount_change: change_amount + amplitude: amplitude + turnover: turnover_rate + timestamp: timestamp + session: session diff --git a/frontend/package.json b/frontend/package.json index 20ff333..fdb5152 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,7 +1,7 @@ { "name": "tickflow-stock-panel-frontend", "private": true, - "version": "0.1.70", + "version": "0.1.80", "type": "module", "scripts": { "dev": "vite", diff --git a/frontend/src/components/Layout.tsx b/frontend/src/components/Layout.tsx index 1473efb..3ff29cb 100644 --- a/frontend/src/components/Layout.tsx +++ b/frontend/src/components/Layout.tsx @@ -281,6 +281,12 @@ export function Layout() { const { data: settingsState } = useSettings() const { data: versionData } = useVersion() const { data: prefs } = usePreferences() + // 数据源列表 (用于实时行情状态显示当前数据源名称) + const { data: dataSources } = useQuery({ + queryKey: QK.dataSources, + queryFn: api.dataSources, + staleTime: 60_000, + }) // poll=true: 全局唯一开启条件轮询 (非交易时段 60s 兜底, 交易时段靠 SSE) const { data: quoteStatus } = useQuoteStatus({ poll: true }) const { data: analysisMenus } = useQuery({ @@ -340,6 +346,21 @@ export function Layout() { const isNoneTier = tier < 0 const isWatchlistMode = tier === 0 const realtimeModeLabel = isWatchlistMode ? '自选股' : '全市场' + // 当前实时行情数据源名称 (custom 时显示源名, tickflow 时不显示) + const realtimeProvider = prefs?.realtime_data_provider + const realtimeProviderName = realtimeProvider && realtimeProvider !== 'tickflow' + ? (dataSources?.custom?.find(s => s.name === realtimeProvider)?.display_name || realtimeProvider) + : null + + // 当前主数据源 (用于菜单底部状态条) + const activeProvider = prefs?.daily_data_provider || 'tickflow' + const activeProviderName = activeProvider === 'tickflow' + ? 'TickFlow' + : (dataSources?.custom?.find(s => s.name === activeProvider)?.display_name || activeProvider) + const activeProviderDatasets = activeProvider === 'tickflow' + ? ['daily', 'adj_factor', 'realtime', 'minute'] + : (dataSources?.custom?.find(s => s.name === activeProvider)?.datasets || []) + const isCustomActive = activeProvider !== 'tickflow' // 轮询触发记录总数 → 更新监控中心徽标 (每 15 秒) const alertsTotalQuery = useQuery({ @@ -476,9 +497,51 @@ export function Layout() { ))} + {/* 数据源状态条 */} + + {/* 全局行情开关 */}
- {isNoneTier ? ( + {isNoneTier && !realtimeProviderName ? (
实时行情 @@ -512,7 +575,7 @@ export function Layout() { : 'bg-muted' }`} /> - 实时行情 · {realtimeModeLabel} + 实时行情 · {realtimeProviderName || realtimeModeLabel}
diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 53e2224..15e8b5c 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -16,7 +16,18 @@ async function request(path: string, init?: RequestInit): Promise { const res = await fetch(`${BASE}${path}`, { ...init, headers }) if (!res.ok) { let detail = '' - try { const j = JSON.parse(await res.text()); detail = j.detail ?? j.message ?? '' } catch { /* ignore */ } + try { + const j = JSON.parse(await res.text()) + const raw = j.detail ?? j.message ?? '' + if (Array.isArray(raw)) { + // FastAPI 422 校验错误: [{type, loc, msg, input}, ...] → 取 msg 拼接 + detail = raw.map((e: any) => e?.msg || String(e)).join('; ') + } else if (typeof raw === 'string') { + detail = raw + } else if (raw && typeof raw === 'object') { + detail = JSON.stringify(raw) + } + } catch { /* ignore */ } const msg = detail || `${res.status} ${res.statusText}` // 401 (未登录/会话过期) 不弹 toast — 由全局认证拦截器统一跳登录页, 避免刷屏 if (res.status !== 401) toast(msg, 'error') @@ -675,6 +686,61 @@ export interface SaveTickflowKeyResult { capabilities_count?: number } +export interface DataSourceItem { + name: string + display_name: string + datasets: string[] + path?: string | null +} + +export interface DataSourceLoadError { + name?: string + path: string + errors: string[] +} + +export interface DataSourcesResponse { + builtin: DataSourceItem[] + custom: DataSourceItem[] + errors: DataSourceLoadError[] + config_dir: string +} + +export interface DataSourceTestResult { + provider: string + dataset: string + rows: number + columns: string[] + preview: Record[] +} + +export interface DatasetConfig { + url: string + method: string + batch?: number | null + rpm?: number | null + response_path: string + field_map: Record + transforms?: Record + symbols_param?: string + start_param?: string + end_param?: string +} + +export interface AuthConfig { + type: string + token_env?: string | null + header?: string + param?: string +} + +export interface CustomSourceConfig { + name: string + display_name: string + auth: AuthConfig + datasets: Record +} + export interface Preferences { realtime_quotes_enabled: boolean indices_nav_pinned: boolean @@ -684,6 +750,7 @@ export interface Preferences { adj_factor_provider?: string minute_data_provider?: string realtime_data_provider?: string + financial_data_provider?: string realtime_watchlist_symbols?: string[] realtime_pull_stock?: boolean realtime_pull_etf?: boolean @@ -781,6 +848,26 @@ export const api = { request<{ ok: boolean }>('/api/settings/ai', { method: 'DELETE' }), preferences: () => request('/api/settings/preferences'), + dataSources: () => request('/api/settings/data-sources'), + dataSource: (name: string) => request(`/api/settings/data-sources/${encodeURIComponent(name)}`), + saveDataSource: (config: CustomSourceConfig) => + request('/api/settings/data-sources', { + method: 'POST', + body: JSON.stringify(config), + }), + deleteDataSource: (name: string) => + request(`/api/settings/data-sources/${encodeURIComponent(name)}`, { method: 'DELETE' }), + reloadDataSources: () => request('/api/settings/data-sources/reload', { method: 'POST' }), + testDataSource: (provider: string, dataset: string, symbols?: string[]) => + request('/api/settings/data-sources/test', { + method: 'POST', + body: JSON.stringify({ provider, dataset, symbols }), + }), + updateDataProviders: (cfg: Partial>) => + request>( + '/api/settings/preferences/data-providers', + { method: 'PUT', body: JSON.stringify(cfg) }, + ), updateMinuteSync: (enabled: boolean, days: number) => request('/api/settings/preferences/minute-sync', { method: 'PUT', diff --git a/frontend/src/lib/queryKeys.ts b/frontend/src/lib/queryKeys.ts index d20c9d5..4dd6fdc 100644 --- a/frontend/src/lib/queryKeys.ts +++ b/frontend/src/lib/queryKeys.ts @@ -14,6 +14,7 @@ export const QK = { endpoints: ['endpoints'] as const, version: ['version'] as const, preferences: ['preferences'] as const, + dataSources: ['data-sources'] as const, quoteStatus: ['quote-status'] as const, quoteInterval: ['quote-interval'] as const, overviewMarket: (asOf?: string) => ['overview-market', asOf ?? 'latest'] as const, diff --git a/frontend/src/pages/Data.tsx b/frontend/src/pages/Data.tsx index 7981453..e710347 100644 --- a/frontend/src/pages/Data.tsx +++ b/frontend/src/pages/Data.tsx @@ -136,6 +136,7 @@ export function Data() { queryKey: QK.extData, queryFn: api.extDataList, }) + const deleteExt = useMutation({ mutationFn: (id: string) => api.extDataDelete(id), onSuccess: () => qc.invalidateQueries({ queryKey: QK.extData }), @@ -152,6 +153,38 @@ export function Data() { }) const prefs = usePreferences() + + // 数据源列表 + 当前数据源 (顶部"切换数据源"按钮展示) + const dataSources = useQuery({ + queryKey: QK.dataSources, + queryFn: api.dataSources, + staleTime: 60_000, + }) + const activeProvider = prefs.data?.daily_data_provider || 'tickflow' + const activeDataSourceName = activeProvider === 'tickflow' + ? 'TickFlow' + : (dataSources.data?.custom?.find(s => s.name === activeProvider)?.display_name || activeProvider) + + // tierKey → 自定义数据集名映射 (用于数据画像 CapBadge 显示数据源名而非 TickFlow 档位) + const TIERKEY_TO_DATASET: Record = { + daily: 'daily', + adj_factor: 'adj_factor', + etf: 'daily', // ETF 复用日K能力 + minute: 'minute', + financials: 'financial', + } + // 当前 custom 源支持的数据集集合 + const activeCustomDatasets = activeProvider !== 'tickflow' + ? new Set(dataSources.data?.custom?.find(s => s.name === activeProvider)?.datasets || []) + : new Set() + // 给定 tierKey, 返回 custom provider 显示名 (走 custom 时) 或 null (走 TickFlow) + const getCustomProviderName = (tierKey: string): string | null => { + if (activeProvider === 'tickflow') return null + const ds = TIERKEY_TO_DATASET[tierKey] + if (ds && activeCustomDatasets.has(ds)) return activeDataSourceName + return null + } + const minuteAuto = prefs.data?.minute_sync_enabled ?? false const pipelineSched = prefs.data?.pipeline_schedule ?? { hour: 15, minute: 30 } const instrumentsSched = prefs.data?.instruments_schedule ?? { hour: 9, minute: 10 } @@ -358,6 +391,7 @@ export function Data() { tierKey="daily" capLimits={caps.data?.capabilities} tierLabel={caps.data?.label} + customProvider={getCustomProviderName('daily')} auto onShowFields={() => setSchemaTable('daily')} onSettings={hasData ? () => setOpenSettings(v => v === 'daily' ? null : 'daily') : undefined} @@ -378,6 +412,7 @@ export function Data() { tierKey="adj_factor" capLimits={caps.data?.capabilities} tierLabel={caps.data?.label} + customProvider={getCustomProviderName('adj_factor')} auto onShowFields={() => setSchemaTable('adj_factor')} /> @@ -440,6 +475,7 @@ export function Data() { tierKey="etf" capLimits={caps.data?.capabilities} tierLabel={caps.data?.label} + customProvider={getCustomProviderName('etf')} auto={etfAuto} subLabel="维表 · 日K · 指标" fieldTabs={[ @@ -464,6 +500,7 @@ export function Data() { tierKey="minute" capLimits={caps.data?.capabilities} tierLabel={caps.data?.label} + customProvider={getCustomProviderName('minute')} auto={minuteAuto} onShowFields={() => setSchemaTable('minute')} onSettings={hasData ? () => setOpenSettings(v => v === 'minute' ? null : 'minute') : undefined} @@ -480,6 +517,7 @@ export function Data() { tierKey="financials" capLimits={caps.data?.capabilities} tierLabel={caps.data?.label} + customProvider={getCustomProviderName('financials')} /> ) default: @@ -540,6 +578,15 @@ export function Data() { 页面设置 +
+ + + {activeDataSourceName} + ))}
diff --git a/frontend/src/pages/settings/DataSourceEditor.tsx b/frontend/src/pages/settings/DataSourceEditor.tsx new file mode 100644 index 0000000..9f766da --- /dev/null +++ b/frontend/src/pages/settings/DataSourceEditor.tsx @@ -0,0 +1,596 @@ +import { useState, useEffect, useRef } from 'react' +import { useMutation, useQuery } from '@tanstack/react-query' +import { motion, AnimatePresence } from 'framer-motion' +import { KeyRound, Play, Plus, Save, Trash2, X, Zap, Check } from 'lucide-react' +import { api, type CustomSourceConfig, type DatasetConfig } from '@/lib/api' +import { toast } from '@/components/Toast' + +// 暗色适配的标准输入框样式 (与 AI 页统一, bg-base 在暗色下为深色, 不会白底白字) +const INPUT_CLS = + 'w-full h-9 px-2.5 rounded-lg bg-base border-0 ring-1 ring-border/40 text-xs text-foreground placeholder:text-muted/30 focus:outline-none focus:ring-2 focus:ring-accent/40 transition-shadow' + +const DATASETS = ['daily', 'adj_factor', 'realtime', 'minute'] as const +type DatasetKey = typeof DATASETS[number] + +const DATASET_LABEL: Record = { + daily: '日K', + adj_factor: '除权因子', + realtime: '实时行情', + minute: '分钟K', +} + +const TARGET_FIELDS: Record = { + daily: ['symbol', 'date', 'open', 'high', 'low', 'close', 'volume', 'amount'], + adj_factor: ['symbol', 'trade_date', 'ex_factor'], + realtime: ['symbol', 'name', 'last_price', 'prev_close', 'open', 'high', 'low', 'volume', 'amount', 'change_pct', 'change_amount', 'amplitude', 'turnover_rate', 'timestamp', 'session'], + minute: ['symbol', 'datetime', 'open', 'high', 'low', 'close', 'volume', 'amount'], +} + +// 内部字段的中文说明 (下拉选项展示用) +const FIELD_LABELS: Record = { + symbol: '股票代码 (如 000001.SZ / 600000.SH)', + date: '交易日期 (YYYY-MM-DD)', + datetime: '时间戳 (YYYY-MM-DD HH:MM:SS)', + open: '开盘价', + high: '最高价', + low: '最低价', + close: '收盘价', + volume: '成交量 (手)', + amount: '成交额 (元)', + trade_date: '除权日期 (YYYY-MM-DD)', + ex_factor: '复权因子', + name: '股票名称', + last_price: '最新价', + prev_close: '昨收价', + change_pct: '涨跌幅 (小数 0.0366=3.66%)', + change_amount: '涨跌额', + amplitude: '振幅 (小数)', + turnover_rate: '换手率 (小数)', + timestamp: '时间戳', + session: '交易时段', +} + +function emptyConfig(): CustomSourceConfig { + return { name: '', display_name: '', auth: { type: 'none' }, datasets: {} } +} + +export function DataSourceEditor({ + existingName, + initial, + onCancel, + onSaved, + activeName, + onActivate, + onDelete, +}: { + existingName?: string + initial?: CustomSourceConfig | null + onCancel: () => void + onSaved: () => void + activeName: string + onActivate: (name: string) => void + onDelete?: () => void +}) { + const isNew = !existingName + const [config, setConfig] = useState(() => initial ? structuredClone(initial) : emptyConfig()) + const [activeTab, setActiveTab] = useState('daily') + + // 编辑现有源: 从后端拉完整配置 (每次挂载都重新拉, 不用缓存, 确保拿到最新保存的配置) + const fetchCfg = useQuery({ + queryKey: ['data-source-detail', existingName], + queryFn: () => api.dataSource(existingName!), + enabled: !!existingName && !initial, + staleTime: 0, + }) + + useEffect(() => { + if (fetchCfg.data) { + setConfig(structuredClone(fetchCfg.data)) + } + }, [fetchCfg.data]) + + const save = useMutation({ + mutationFn: () => { + // 提交前校验: 每个已启用数据集必须填了 URL + for (const [key, ds] of Object.entries(config.datasets)) { + if (!ds.url.trim()) { + throw new Error(`数据集「${DATASET_LABEL[key as DatasetKey] || key}」未填写接口 URL`) + } + } + // 提交时去掉 field_map 里的 __pending_ 临时 key (未填外部字段名的草稿行) + const cleaned: CustomSourceConfig = { + ...config, + name: config.name.toLowerCase().trim(), + display_name: config.display_name.trim() || config.name.toLowerCase().trim(), + datasets: Object.fromEntries( + Object.entries(config.datasets).map(([k, ds]) => [ + k, + { + ...ds, + field_map: Object.fromEntries( + Object.entries(ds.field_map).filter(([src]) => !src.startsWith('__pending_')) + ), + }, + ]) + ), + } + return api.saveDataSource(cleaned) + }, + onSuccess: () => { + toast(isNew ? '数据源已创建' : '数据源已更新', 'success') + // 保存后强制重新拉取最新配置, 让数据集开关状态正确刷新 + fetchCfg.refetch() + onSaved() + }, + onError: (e: Error) => { + toast(e.message, 'error') + }, + }) + + const setDatasetEnabled = (key: DatasetKey, enabled: boolean) => { + setConfig(prev => { + const next = { ...prev, datasets: { ...prev.datasets } } + if (enabled) { + if (!next.datasets[key]) { + next.datasets[key] = { url: '', method: 'POST', response_path: 'data', field_map: {} } + } + } else { + delete next.datasets[key] + } + return next + }) + } + + const updateDataset = (key: DatasetKey, patch: Partial) => { + setConfig(prev => ({ + ...prev, + datasets: { ...prev.datasets, [key]: { ...prev.datasets[key], ...patch } as DatasetConfig }, + })) + } + + const canSave = !!config.name.trim() && !save.isPending + const loading = !!existingName && !initial && fetchCfg.isLoading + const isActive = !isNew && activeName === existingName + + return ( +
+ {/* 头部 */} +
+
+
+ {isNew ? : } +
+
+

{isNew ? '新增数据源' : '编辑数据源'}

+

{isNew ? '配置一个自定义 HTTP 数据源' : config.display_name || existingName}

+
+
+
+ {!isNew && isActive && ( + + 使用中 + + )} + {!isNew && !isActive && config.name.trim() && ( + + )} + {!isNew && onDelete && ( + + )} +
+
+ + {loading ? ( +
加载配置...
+ ) : ( +
+ {/* 左: 基本信息 + 鉴权 + 数据集开关 */} +
+ + setConfig({ ...config, name: e.target.value })} + placeholder="my_tushare" + disabled={!isNew} + className={`${INPUT_CLS} w-full disabled:opacity-60`} + /> + + + setConfig({ ...config, display_name: e.target.value })} + placeholder="我的 Tushare" + className={`${INPUT_CLS} w-full`} + /> + + +
+ + {config.auth.type !== 'none' && ( + setConfig({ ...config, auth: { ...config.auth, token_env: e.target.value } })} + placeholder="环境变量名 (MY_TOKEN)" + className={`${INPUT_CLS} w-full`} + /> + )} +
+ + +
+
数据集
+ {DATASETS.map(key => { + const enabled = !!config.datasets[key] + return ( + + ) + })} +
+
+ + {/* 右: 当前数据集详情 */} +
+ updateDataset(activeTab, patch)} + onFieldMap={(fm) => updateDataset(activeTab, { field_map: fm })} + onToggle={(v) => setDatasetEnabled(activeTab, v)} + /> +
+
+ )} + + {/* 底部保存栏 */} +
+
+ {Object.keys(config.datasets).length} 个数据集已配置 +
+
+ + +
+
+
+ ) +} + +function DatasetDetail({ + datasetKey, + cfg, + providerName, + onUpdate, + onFieldMap, + onToggle, +}: { + datasetKey: DatasetKey + cfg?: DatasetConfig + providerName: string + onUpdate: (patch: Partial) => void + onFieldMap: (fm: Record) => void + onToggle: (v: boolean) => void +}) { + const enabled = !!cfg + const [testSymbols, setTestSymbols] = useState('000001.SZ,600000.SH') + const test = useMutation({ + mutationFn: () => api.testDataSource( + providerName, + datasetKey, + testSymbols.split(/[,\s]+/).map(s => s.trim()).filter(Boolean), + ), + }) + + return ( +
+
+
+

{DATASET_LABEL[datasetKey]}

+ {datasetKey} +
+ onToggle(!enabled)} /> +
+ + + {enabled && cfg ? ( + +
+ + onUpdate({ url: e.target.value })} + placeholder="https://my.api/daily" + className={`${INPUT_CLS} w-full`} + /> + + + + +
+ +
+ + onUpdate({ batch: e.target.value ? Number(e.target.value) : null })} + placeholder="100" + className={`${INPUT_CLS} w-full`} + /> + + + onUpdate({ rpm: e.target.value ? Number(e.target.value) : null })} + placeholder="200" + className={`${INPUT_CLS} w-full`} + /> + + + onUpdate({ response_path: e.target.value })} + placeholder="data.list" + className={`${INPUT_CLS} w-full`} + /> + +
+ +
+
+
字段映射
+
外部字段 → 内部字段
+
+ +
+ +
+
+ + 测试连接 +
+
+ setTestSymbols(e.target.value)} + className={`${INPUT_CLS} flex-1 text-xs`} + placeholder="测试标的, 逗号分隔" + /> + +
+ {test.data && ( +
+ {test.data.rows} 行 + · + 列: {test.data.columns.join(', ')} +
+ )} + {test.isError && ( +
测试失败, 请检查接口和映射
+ )} +
+ + ) : ( + +
{DATASET_LABEL[datasetKey]} 未启用
+
启用后此数据集将由该自定义源提供, 未启用则回退 TickFlow
+ + + )} + +
+ ) +} + +function FieldMapEditor({ + fieldMap, + targets, + onChange, +}: { + fieldMap: Record + targets: string[] + onChange: (fm: Record) => void +}) { + // 内部用数组维护行的稳定身份, 避免 Record 在编辑空行时 key 漂移导致输入框失焦 + const [rows, setRows] = useState>(() => { + const entries = Object.entries(fieldMap) + // 有意义的映射 (src 非空且非 pending) + const real = entries.filter(([s, t]) => s.trim() && t.trim() && !s.startsWith('__pending_')) + // pending 行 (外部字段名还没填, 但 target 已选) + const pending = entries.filter(([s]) => s.startsWith('__pending_')) + if (real.length > 0 || pending.length > 0) { + return [ + ...real.map(([src, target], i) => ({ src, target, id: i + 1 })), + ...pending.map(([, target], i) => ({ src: '', target, id: real.length + i + 1 })), + ] + } + // fieldMap 为空时自动预填该数据集的所有内部字段 (外部字段名留空待填) + return targets.map((target, i) => ({ src: '', target, id: i + 1 })) + }) + const nextId = useRef(targets.length + 1) + + // rows 变化时立即同步到父级 (含空 src 的草稿行, 用 __pending_ 前缀保留) + // 这样切换 tab 再切回来, 未填完的映射行不会丢 + useEffect(() => { + const out: Record = {} + let pendingIdx = 0 + for (const r of rows) { + const s = r.src.trim() + if (s && r.target.trim()) { + out[s] = r.target.trim() + } else if (r.target.trim()) { + // 外部字段名还没填, 用临时 key 保留 target 选择 + out[`__pending_${pendingIdx++}`] = r.target.trim() + } + } + onChange(out) + }, [rows]) // eslint-disable-line react-hooks/exhaustive-deps + + const emit = (newRows: typeof rows) => { + setRows(newRows) + } + + const updateRow = (id: number, patch: Partial<{ src: string; target: string }>) => { + emit(rows.map(r => (r.id === id ? { ...r, ...patch } : r))) + } + + const removeRow = (id: number) => { + const filtered = rows.filter(r => r.id !== id) + emit(filtered.length > 0 ? filtered : [{ src: '', target: '', id: nextId.current++ }]) + } + + const addRow = () => { + emit([...rows, { src: '', target: '', id: nextId.current++ }]) + } + + const hasValid = rows.some(r => r.src.trim() && r.target.trim()) + + return ( +
+ {!hasValid && ( +
填写外部字段名后自动生效, 无需的字段可删除
+ )} + {rows.map((row) => ( +
+ updateRow(row.id, { src: e.target.value })} + placeholder="外部字段名" + className={`${INPUT_CLS} text-xs`} + /> + → + + +
+ ))} + +
+ ) +} + +function Field({ label, hint, children }: { label: string; hint?: string; children: React.ReactNode }) { + return ( +
+
+ {label} + {hint && {hint}} +
+ {children} +
+ ) +} + +function Toggle({ checked, onChange }: { checked: boolean; onChange: (e?: React.MouseEvent) => void }) { + return ( + + ) +} diff --git a/frontend/src/pages/settings/DataSources.tsx b/frontend/src/pages/settings/DataSources.tsx new file mode 100644 index 0000000..163eb31 --- /dev/null +++ b/frontend/src/pages/settings/DataSources.tsx @@ -0,0 +1,331 @@ +import { useState } from 'react' +import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query' +import { motion, AnimatePresence } from 'framer-motion' +import { Check, Database, Plus, RefreshCw, Zap, FileWarning } from 'lucide-react' +import { api, type DataSourceItem } from '@/lib/api' +import { QK } from '@/lib/queryKeys' +import { usePreferences } from '@/lib/useSharedQueries' +import { toast } from '@/components/Toast' +import { DataSourceEditor } from './DataSourceEditor' + +const DATASET_LABEL: Record = { + daily: '日K', + adj_factor: '除权', + realtime: '实时', + minute: '分钟', +} + +export function SettingsDataSourcesPanel() { + const qc = useQueryClient() + const prefs = usePreferences() + const sources = useQuery({ queryKey: QK.dataSources, queryFn: api.dataSources }) + const [selected, setSelected] = useState('tickflow') // 当前在右侧编辑的源 name + const [confirmDelete, setConfirmDelete] = useState(null) + + const reload = useMutation({ + mutationFn: api.reloadDataSources, + onSuccess: () => { + qc.invalidateQueries({ queryKey: QK.dataSources }) + toast('配置已重新加载', 'success') + }, + }) + + const remove = useMutation({ + mutationFn: (name: string) => api.deleteDataSource(name), + onSuccess: () => { + qc.invalidateQueries({ queryKey: QK.dataSources }) + qc.invalidateQueries({ queryKey: QK.preferences }) + setSelected('tickflow') + setConfirmDelete(null) + toast('数据源已删除', 'success') + }, + }) + + const switchProvider = useMutation({ + mutationFn: (name: string) => { + if (name === 'tickflow') { + return api.updateDataProviders({ + daily_data_provider: 'tickflow', + adj_factor_provider: 'same_as_daily', + realtime_data_provider: 'tickflow', + minute_data_provider: 'tickflow', + financial_data_provider: 'tickflow', + }) + } + return api.updateDataProviders({ + daily_data_provider: name, + adj_factor_provider: 'same_as_daily', + realtime_data_provider: name, + financial_data_provider: name, + }) + }, + onSuccess: () => { + qc.invalidateQueries({ queryKey: QK.preferences }) + toast('数据源已切换', 'success') + }, + }) + + const editExisting = useMutation({ + mutationFn: (name: string) => api.dataSource(name), + onSuccess: (_data, name) => setSelected(name), + }) + + const builtin: DataSourceItem[] = sources.data?.builtin ?? [] + const customList: DataSourceItem[] = sources.data?.custom ?? [] + const errors = sources.data?.errors ?? [] + const activeName = prefs.data?.daily_data_provider || 'tickflow' + + // 顶部数据源选择列表 (内置 + 自定义 + 新增) + const allItems = [ + ...builtin, + ...customList, + ] + + const selectedCustom = customList.find(s => s.name === selected) + + return ( +
+ {/* ===== 顶部: 当前数据源 + 数据源选择 (一个大卡片) ===== */} +
+
+
+ +

数据源

+ + {sources.data?.config_dir} + +
+ +
+ + {/* 当前数据源状态 */} +
+ 当前 + + + {activeName === 'tickflow' ? 'TickFlow' : customList.find(s => s.name === activeName)?.display_name || activeName} + +
+ + {/* 数据源选择 - 横向卡片列表 */} +
+ {allItems.map(item => { + const isActive = activeName === item.name + const isSelected = selected === item.name + return ( +
{ + setSelected(item.name) + if (item.name !== 'tickflow') { + editExisting.mutate(item.name) + } + }} + className={`relative cursor-pointer text-left rounded-lg border px-3.5 py-3 transition-all ${ + isSelected + ? 'border-accent/50 bg-accent/5 ring-1 ring-accent/20' + : 'border-border/60 bg-elevated/20 hover:bg-elevated/40' + }`} + > +
+ + + {item.display_name} + + {item.name === 'tickflow' && ( + 内置 + )} + {isActive ? ( + + 使用中 + + ) : ( + + )} +
+ {item.name !== 'tickflow' && item.datasets.length > 0 && ( +
+ {item.datasets.map(ds => ( + + {DATASET_LABEL[ds] || ds} + + ))} +
+ )} + {item.name === 'tickflow' && ( +
日K · 除权 · 实时 · 分钟K
+ )} +
+ ) + })} + + {/* 新增数据源卡片 */} + +
+ + {/* 错误提示 */} + {errors.length > 0 && ( +
+ +
+ {errors.map((err, idx) => ( +
+ {err.name || err.path}: {err.errors.join('; ')} +
+ ))} +
+
+ )} + +
+ 单击编辑 + · + 点「使用」切换为当前数据源 + · + 未启用的数据集自动回退 TickFlow +
+
+ + {/* ===== 下方: 编辑区 ===== */} + + + {selected === 'tickflow' ? ( + switchProvider.mutate('tickflow')} + switching={switchProvider.isPending} + /> + ) : ( + setSelected('tickflow')} + onSaved={() => { + qc.invalidateQueries({ queryKey: QK.dataSources }) + // 强制清除该源的详情缓存, 下次编辑重新拉取最新配置 + if (selected !== '__new__') { + qc.removeQueries({ queryKey: ['data-source-detail', selected] }) + } + if (selected === '__new__') setSelected(activeName === 'tickflow' ? 'tickflow' : activeName) + }} + activeName={activeName} + onActivate={(name) => switchProvider.mutate(name)} + onDelete={selected !== '__new__' && selectedCustom ? () => setConfirmDelete(selected) : undefined} + /> + )} + + + + {/* 删除确认弹窗 */} + {confirmDelete && ( +
+
setConfirmDelete(null)} + /> +
+

删除数据源

+

+ 确认删除「{customList.find(s => s.name === confirmDelete)?.display_name || confirmDelete}」? 该数据源的配置文件将被移除,此操作不可撤销。 +

+
+ + +
+
+
+ )} +
+ ) +} + +function TickFlowDetail({ active, onSwitch, switching }: { active: boolean; onSwitch: () => void; switching: boolean }) { + return ( +
+
+
+ +
+
+
+

TickFlow

+ 内置默认 + {active && ( + + 当前使用 + + )} +
+

+ 项目默认数据源。日K、除权因子、实时行情、分钟K均由 TickFlow 提供,无需额外配置。 +

+
+
+ +
+ {[ + { label: '日K', desc: '历史 + 实时覆写' }, + { label: '除权因子', desc: 'Starter+ 能力' }, + { label: '实时行情', desc: '全市场快照' }, + { label: '分钟K', desc: 'Pro+ 能力' }, + ].map(f => ( +
+
{f.label}
+
{f.desc}
+
+ ))} +
+ + {!active && ( + + )} +
+ ) +}