127 lines
4.3 KiB
Python
127 lines
4.3 KiB
Python
"""TickFlow SDK 封装(§5)。
|
|
|
|
进程内单例;Key 来源(优先级):secrets.json > .env。
|
|
用户改 Key 后需要 `reset_clients()`,然后 `get_client()` 会拿新的。
|
|
|
|
5 档体系下服务器归属:
|
|
- none 档(无 key / 无效 key) → TickFlow.free()(free-api 服务器)
|
|
- free 档(免费有效 key) → TickFlow.free()(key 被 SDK 忽略,运行时走 free-api)
|
|
- starter/pro/expert(付费 key) → TickFlow(api_key=key, base_url)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
from tickflow import AsyncTickFlow, TickFlow
|
|
|
|
from app import secrets_store
|
|
|
|
_sync_client: TickFlow | None = None
|
|
_async_client: AsyncTickFlow | None = None
|
|
_paid_realtime_client: TickFlow | None = None
|
|
|
|
|
|
# ===== 服务器归属判定 =====
|
|
|
|
# free-api 服务器默认节点(SDK 默认值),none/free 档运行时走这里。
|
|
FREE_ENDPOINT = "https://free-api.tickflow.org"
|
|
# 付费端点默认节点(starter+ 运行时走这里,也是端点切换的默认值)。
|
|
PAID_ENDPOINT = "https://api.tickflow.org"
|
|
|
|
|
|
def _should_use_free_server() -> bool:
|
|
"""是否应走 free-api 服务器。
|
|
|
|
判定依据:无 key,或当前档位为 none/free。
|
|
付费档(starter+)走付费端点。
|
|
"""
|
|
if not secrets_store.get_tickflow_key():
|
|
return True
|
|
# 有 key 时按探测出的档位判定(避免读 capabilities.json 在首次启动前未生成的边界)
|
|
from app.tickflow.policy import base_tier_name
|
|
return base_tier_name() in ("none", "free")
|
|
|
|
|
|
def _base_url() -> str | None:
|
|
"""从 secrets.json 读取用户自定义端点,没有则返回 None(用 SDK 默认)。"""
|
|
return secrets_store.load().get("tickflow_base_url") or None
|
|
|
|
|
|
def get_client() -> TickFlow:
|
|
"""同步客户端。能力探测、盘后管道用。"""
|
|
global _sync_client
|
|
if _sync_client is None:
|
|
key = secrets_store.get_tickflow_key()
|
|
if _should_use_free_server():
|
|
# none/free 档:走 free-api 服务器(无 key 或免费 key 被 SDK 忽略)
|
|
_sync_client = TickFlow.free()
|
|
else:
|
|
_sync_client = TickFlow(api_key=key, base_url=_base_url())
|
|
return _sync_client
|
|
|
|
|
|
def get_async_client() -> AsyncTickFlow:
|
|
"""异步客户端。FastAPI 请求路径上用。"""
|
|
global _async_client
|
|
if _async_client is None:
|
|
key = secrets_store.get_tickflow_key()
|
|
if _should_use_free_server():
|
|
_async_client = AsyncTickFlow.free()
|
|
else:
|
|
_async_client = AsyncTickFlow(api_key=key, base_url=_base_url())
|
|
return _async_client
|
|
|
|
|
|
def get_paid_realtime_client() -> TickFlow | None:
|
|
"""实时行情专用付费服务器客户端。
|
|
|
|
none/free 的历史日K仍走 get_client() 的 free-api；实时行情全部走付费服务器。
|
|
Free 档如果有有效 key，也使用这里的 paid endpoint 调按标的实时接口。
|
|
"""
|
|
global _paid_realtime_client
|
|
key = secrets_store.get_tickflow_key()
|
|
if not key:
|
|
return None
|
|
if _paid_realtime_client is None:
|
|
_paid_realtime_client = TickFlow(api_key=key, base_url=_base_url())
|
|
return _paid_realtime_client
|
|
|
|
|
|
def reset_clients() -> None:
|
|
"""Key 变化后调用 — 让下一次 get_client() 拿新实例。"""
|
|
global _sync_client, _async_client, _paid_realtime_client
|
|
_sync_client = None
|
|
_async_client = None
|
|
_paid_realtime_client = None
|
|
|
|
|
|
def current_mode() -> str:
|
|
"""供 UI 显示当前模式。三态:
|
|
|
|
- "none" : 无 key / 无效 key(走 free-api,仅历史日K)
|
|
- "free" : 免费有效 key(走 free-api,仅历史日K)
|
|
- "api_key" : 付费 key(starter+,走付费端点,有实时行情)
|
|
"""
|
|
if not secrets_store.get_tickflow_key():
|
|
return "none"
|
|
from app.tickflow.policy import base_tier_name
|
|
tier = base_tier_name()
|
|
if tier in ("none", "free"):
|
|
return "free" if tier == "free" else "none"
|
|
return "api_key"
|
|
|
|
|
|
def current_endpoint() -> str:
|
|
"""返回当前显示用的端点 URL(对应 endpoints.json 列表项)。
|
|
|
|
- none/free 档:显示 free-api 服务器节点
|
|
- 付费档:显示用户自定义端点(测速切换后)或默认付费节点 api.tickflow.org
|
|
"""
|
|
if _should_use_free_server():
|
|
return FREE_ENDPOINT
|
|
# 自定义端点(付费模式测速切换后):优先返回
|
|
base = _base_url()
|
|
if base:
|
|
return base.rstrip("/")
|
|
return PAID_ENDPOINT
|