37 lines
1.3 KiB
Python
37 lines
1.3 KiB
Python
from dataclasses import dataclass, field
|
|
import secrets
|
|
|
|
from .http import get_json
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class SignalItem:
|
|
signal_key: str = ""; code: str = ""; name: str = ""; desc: str = ""; last_close: float = 0
|
|
tech_indicator: dict[str, float] = field(default_factory=dict)
|
|
|
|
@dataclass(slots=True)
|
|
class SignalResult:
|
|
code: str = ""; total: int = 0; updated: str = ""; data: dict[str, SignalItem] = field(default_factory=dict); message: str = ""
|
|
|
|
|
|
def fetch_signal(api_host: str, sub_url: str, timeout: float = 5.0) -> SignalResult:
|
|
url = f"{api_host}{sub_url}?t={secrets.token_urlsafe(12)}"
|
|
try:
|
|
raw = get_json(url, timeout)
|
|
except Exception:
|
|
return SignalResult()
|
|
items = {code: SignalItem(**item) for code, item in (raw.get("data") or {}).items()}
|
|
return SignalResult(raw.get("code", ""), raw.get("total", 0), raw.get("updated", ""), items, raw.get("message", ""))
|
|
|
|
|
|
def init_signals(global_config, allow: list[str]) -> list[SignalItem]:
|
|
result = []
|
|
for key, cfg in global_config.signals.items():
|
|
if key not in allow:
|
|
continue
|
|
signal_result = fetch_signal(global_config.api_host, cfg.url)
|
|
for item in signal_result.data.values():
|
|
item.signal_key = key
|
|
result.append(item)
|
|
return result
|