This commit is contained in:
qq564020069
2026-07-14 19:08:12 +08:00
parent 3728825adc
commit 4cf76ce51f
18 changed files with 877 additions and 157 deletions
+34
View File
@@ -46,6 +46,32 @@ def _strategy_file_for_symbol(symbol: str | None) -> str:
return _STRATEGY_FILE
def _fallback_data_file_for_symbol(symbol: str) -> tuple[str | None, str | None]:
"""Read web_settings.json last_data_file when strategy JSON lacks data_file."""
settings_path = pathlib.Path("web_settings.json")
if not settings_path.exists():
return None, None
try:
settings = json.loads(settings_path.read_text(encoding="utf-8"))
last = str(settings.get("last_data_file") or "").strip()
except (json.JSONDecodeError, OSError):
return None, None
if not last:
return None, None
p = pathlib.Path(last)
if not p.exists():
return None, None
try:
from data_pipeline.parquet_manager import inspect_parquet_file
info = inspect_parquet_file(str(p.resolve()))
except Exception:
return str(p.resolve()), None
if info.get("symbol") != symbol:
return None, None
return str(p.resolve()), info.get("timeframe")
# ─────────────────────────────────────────────────────────────────────────────
# Walk-Forward 折叠构建
# ─────────────────────────────────────────────────────────────────────────────
@@ -1007,6 +1033,14 @@ class AlphaEngine:
val = existing.get(key)
if val is not None:
strategy_data[key] = val
if not strategy_data.get("data_file") and self.target_symbol:
data_file, tf = _fallback_data_file_for_symbol(self.target_symbol)
if data_file:
strategy_data["data_file"] = data_file
if tf and not strategy_data.get("timeframe"):
strategy_data["timeframe"] = tf
if data_file and not strategy_data.get("mode"):
strategy_data["mode"] = "parquet_file"
with open(save_path, "w", encoding="utf-8") as fp:
json.dump(strategy_data, fp, indent=2, ensure_ascii=False)
+3
View File
@@ -9,3 +9,6 @@ seaborn>=0.12.0
# Chinese Stock Market Data (for times.py only)
tushare>=1.2.89
# Tongdaxin realtime quotes (Web UI data source)
pytdx>=1.72
+11
View File
@@ -0,0 +1,11 @@
"""Shared pytest fixtures."""
from __future__ import annotations
import pytest
@pytest.fixture(autouse=True)
def isolate_web_settings(tmp_path, monkeypatch):
"""Never let unit tests overwrite the real web_settings.json."""
settings_path = tmp_path / "web_settings.json"
monkeypatch.setattr("web.settings.SETTINGS_PATH", settings_path)
+58
View File
@@ -0,0 +1,58 @@
"""Recover invalid last_data_file paths in web settings."""
from __future__ import annotations
import json
from pathlib import Path
import pytest
import web.settings as settings_mod
from web.settings import load_settings, save_settings
@pytest.fixture
def project(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
strategies = tmp_path / "strategies"
strategies.mkdir()
monkeypatch.setattr(settings_mod, "STRATEGIES_DIR", strategies)
return tmp_path
def test_load_settings_recovers_missing_temp_path(project: Path) -> None:
parquet = project / "XAUUSD_H1.parquet"
parquet.write_bytes(b"PAR1")
strat = project / "strategies" / "best_XAUUSD.json"
strat.write_text(
json.dumps({"symbol": "XAUUSD", "data_file": str(parquet.resolve())}),
encoding="utf-8",
)
settings_path = settings_mod.SETTINGS_PATH
settings_path.write_text(
json.dumps(
{
"last_data_file": r"C:\Users\x\AppData\Local\Temp\2\pytest-of-x\test0\XAUUSD_H1.parquet",
"last_strategy_file": str(strat.resolve()),
}
),
encoding="utf-8",
)
loaded = load_settings()
assert loaded["last_data_file"] == str(parquet.resolve())
if settings_mod._is_production_settings_path():
persisted = json.loads(settings_path.read_text(encoding="utf-8"))
assert persisted["last_data_file"] == str(parquet.resolve())
def test_save_settings_ignores_ephemeral_path(project: Path, monkeypatch: pytest.MonkeyPatch) -> None:
prod_settings = project / "web_settings.json"
monkeypatch.setattr(settings_mod, "SETTINGS_PATH", prod_settings)
monkeypatch.setattr(settings_mod, "PROJECT_ROOT", project)
save_settings({"last_data_file": "D:\\real\\XAUUSD_H1.parquet"})
save_settings(
{
"last_data_file": r"C:\Users\x\AppData\Local\Temp\2\pytest-of-x\test0\X.parquet"
}
)
loaded = load_settings()
assert loaded["last_data_file"] == "D:\\real\\XAUUSD_H1.parquet"
+76
View File
@@ -0,0 +1,76 @@
"""Strategy data_file fallback from training settings."""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from web import strategy_file as sf
from web.settings import save_settings
@pytest.fixture
def project(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
strategies = tmp_path / "strategies"
strategies.mkdir()
settings_path = tmp_path / "web_settings.json"
import web.progress as progress_mod
monkeypatch.setattr(progress_mod, "STRATEGIES_DIR", strategies)
monkeypatch.setattr(sf, "STRATEGIES_DIR", strategies)
monkeypatch.setattr(sf, "checkpoint_glob", lambda _symbol: [])
import web.settings as settings_mod
monkeypatch.setattr(settings_mod, "SETTINGS_PATH", settings_path)
monkeypatch.setattr(settings_mod, "STRATEGIES_DIR", strategies)
monkeypatch.chdir(tmp_path)
def fake_inspect(path: str) -> dict:
return {"symbol": "XAUUSD", "timeframe": "H1", "bars": 100, "valid": True}
monkeypatch.setattr(sf, "inspect_parquet_file", fake_inspect)
return tmp_path
def test_inspect_strategy_fills_data_file_from_settings(project: Path) -> None:
parquet = project / "XAUUSD_H1.parquet"
parquet.write_bytes(b"PAR1")
save_settings({"last_data_file": str(parquet.resolve())})
strat_path = project / "strategies" / "best_XAUUSD.json"
strat_path.write_text(
json.dumps({"symbol": "XAUUSD", "formula": [1, 2, 3], "best_score": 1.5}),
encoding="utf-8",
)
info = sf.inspect_strategy_file(str(strat_path))
assert info["data_file"] == str(parquet.resolve())
assert info["data_file_exists"] is True
assert info["timeframe"] == "H1"
def test_sync_best_writes_data_file(project: Path) -> None:
parquet = project / "XAUUSD_H1.parquet"
parquet.write_bytes(b"PAR1")
save_settings({"last_data_file": str(parquet.resolve())})
strat_path = project / "strategies" / "best_XAUUSD.json"
strat_path.write_text(
json.dumps(
{
"symbol": "XAUUSD",
"formula": [1, 2, 3],
"best_score": 1.5,
"formula_decoded": "A → B",
}
),
encoding="utf-8",
)
info = sf.sync_best_strategy_for_symbol("XAUUSD")
assert info is not None
assert info["data_file"] == str(parquet.resolve())
saved = json.loads(strat_path.read_text(encoding="utf-8"))
assert saved["data_file"] == str(parquet.resolve())
+15
View File
@@ -144,6 +144,21 @@ def _save_strategy(engine: AlphaEngine, symbol: str, timeframe: str, data_file:
f" [策略] 保留磁盘更优结果 {float(old_score):.4f} "
f"> 本次 {float(engine.best_score):.4f},未覆盖 {path}"
)
merged = dict(old)
for key, val in (
("timeframe", timeframe),
("data_file", str(Path(data_file).resolve())),
("mode", "parquet_file"),
("train_steps", ModelConfig.TRAIN_STEPS),
):
if val is not None and not merged.get(key):
merged[key] = val
if merged != old:
path.write_text(
json.dumps(merged, indent=2, ensure_ascii=False),
encoding="utf-8",
)
print(f" [策略] 已补全数据路径等元数据: {path}")
return
except (json.JSONDecodeError, OSError, TypeError, ValueError):
pass
+40 -13
View File
@@ -85,7 +85,7 @@ def detect_qclaw(*, require_alive: bool = False) -> bool:
return True
host, port, token = info
base = f"http://{host}:{port}/v1"
return _pick_openclaw_model(base, token, probe_only=True)
return _probe_qclaw_gateway(base, token)
def detect_workbuddy() -> bool:
@@ -98,15 +98,42 @@ def detect_workbuddy() -> bool:
return _WORKBUDDY_CONFIG_DIR.exists()
def _probe_qclaw_gateway(base_url: str, token: str, *, timeout: float = 5.0) -> bool:
"""Return True when the local QClaw gateway responds."""
import urllib.error
import urllib.request
headers = {"Authorization": f"Bearer {token}", "User-Agent": "AlphaMaster"}
for path in ("/models", "/health"):
try:
req = urllib.request.Request(
f"{base_url.rstrip('/')}{path}",
headers=headers,
method="GET",
)
with urllib.request.urlopen(req, timeout=timeout) as resp:
if resp.status == 200:
return True
except urllib.error.HTTPError as exc:
if exc.code in (401, 403):
return True
except Exception as exc:
logger.debug("QClaw probe %s failed: %s", path, exc)
return False
def provider_status() -> dict[str, Any]:
qclaw_cfg = detect_qclaw(require_alive=False)
qclaw_ok = detect_qclaw(require_alive=True) if qclaw_cfg else False
qclaw_reachable = detect_qclaw(require_alive=True) if qclaw_cfg else False
wb_ok = bool(_workbuddy_token())
wb_env = detect_workbuddy()
if qclaw_ok:
if qclaw_cfg and qclaw_reachable:
qclaw_hint = "已检测到本地 QClaw Gateway(可连通)"
elif qclaw_cfg:
qclaw_hint = "已找到 QClaw 配置,但 Gateway 未启动(请先打开 QClaw)"
qclaw_hint = (
"已读取 QClaw 配置,将自动连接本地 Gateway。"
"若 QClaw 已打开仍显示未连通,可直接尝试分析(分析时会重新检测)。"
)
else:
qclaw_hint = "未检测到 QClaw(需 ~/.qclaw/openclaw.json 且 chatCompletions 已启用)"
if wb_ok:
@@ -131,7 +158,8 @@ def provider_status() -> dict[str, Any]:
{
"id": "openclaw",
"label": "openclaw (QClaw)",
"available": qclaw_ok,
"available": qclaw_cfg,
"gateway_reachable": qclaw_reachable,
"needs_user_key": False,
"hint": qclaw_hint,
},
@@ -193,9 +221,10 @@ def resolve_provider(provider: str, api_key: str | None = None) -> ResolvedProvi
)
host, port, token = info
base = f"http://{host}:{port}/v1"
if not _pick_openclaw_model(base, token, probe_only=True):
if not _probe_qclaw_gateway(base, token):
raise ValueError(
f"已读取 QClaw token,但 Gateway 未连通({base})。请先启动 QClaw 后再试。"
f"已读取 QClaw 配置,但暂时无法连接 Gateway({base})。"
"请确认 QClaw 已打开;若刚启动请稍等几秒后重试。"
)
model = str(_pick_openclaw_model(base, token) or _OPENCLAW_MODEL)
return ResolvedProvider(
@@ -352,6 +381,8 @@ def _pick_openclaw_model(
When *probe_only* is True, return whether the gateway responded.
Otherwise return a preferred model id (falls back to ``openclaw``).
"""
if probe_only:
return _probe_qclaw_gateway(base_url, token)
try:
import urllib.request
@@ -360,10 +391,8 @@ def _pick_openclaw_model(
headers={"Authorization": f"Bearer {token}", "User-Agent": "AlphaMaster"},
method="GET",
)
with urllib.request.urlopen(req, timeout=2.0) as resp:
with urllib.request.urlopen(req, timeout=5.0) as resp:
data = json.loads(resp.read().decode("utf-8"))
if probe_only:
return True
ids = [str(m.get("id", "")) for m in data.get("data", []) if m.get("id")]
if _OPENCLAW_MODEL in ids:
return _OPENCLAW_MODEL
@@ -372,9 +401,7 @@ def _pick_openclaw_model(
return mid
except Exception as exc:
logger.debug("QClaw /models probe failed: %s", exc)
if probe_only:
return False
return False if probe_only else _OPENCLAW_MODEL
return _OPENCLAW_MODEL
def _workbuddy_auth_dirs() -> list[Path]:
+44 -4
View File
@@ -208,7 +208,10 @@ def _strategy_context() -> dict[str, Any]:
strategy_info = None
if resolved:
try:
strategy_info = inspect_strategy_file(resolved)
strategy_info = inspect_strategy_file(
resolved,
data_file_hint=settings.get("last_data_file") or None,
)
except Exception as e:
strategy_info = {
"strategy_file": resolved,
@@ -276,9 +279,20 @@ def _wait_training_idle(timeout_s: float = 5.0) -> None:
time.sleep(0.2)
def _sync_and_persist_best_strategy(symbol: str) -> dict[str, Any] | None:
def _sync_and_persist_best_strategy(
symbol: str,
*,
data_file_hint: str | None = None,
) -> dict[str, Any] | None:
invalidate_checkpoint_cache()
info = sync_best_strategy_for_symbol(symbol)
hint = data_file_hint
if not hint:
job = training_manager.status().get("job") or {}
if str(job.get("symbol") or "") == symbol:
hint = job.get("data_file") or None
if not hint:
hint = load_settings().get("last_data_file") or None
info = sync_best_strategy_for_symbol(symbol, data_file_hint=hint)
if info:
save_settings({"last_strategy_file": info["strategy_file"]})
return info
@@ -693,11 +707,15 @@ def api_training_start(req: StartTrainingRequest) -> dict[str, Any]:
def api_training_stop() -> dict[str, Any]:
job = training_manager.status().get("job") or {}
symbol = job.get("symbol")
data_file_hint = job.get("data_file")
stopped = training_manager.stop()
strategy_file = None
if symbol:
_wait_training_idle()
strategy_file = _sync_and_persist_best_strategy(symbol)
strategy_file = _sync_and_persist_best_strategy(
symbol,
data_file_hint=data_file_hint,
)
return {
"ok": stopped,
"training": training_manager.status(),
@@ -953,6 +971,28 @@ def api_realtime_sources() -> dict[str, Any]:
return {"sources": list_sources(), "min_exposure": min_exposure()}
@app.post("/api/realtime/tradingview/probe")
def api_realtime_tradingview_probe() -> dict[str, Any]:
"""Probe TradingView reachability (same behavior as PA_Agent before fetch)."""
from web.data_sources.tradingview_connectivity import (
TV_CLOUD_SERVER_WIKI_URL,
TV_CONNECTIVITY_MESSAGE,
check_tradingview_connectivity,
)
ok, detail = check_tradingview_connectivity(
timeout_s=15.0, max_attempts=2, retry_delay_s=2.0
)
return {
"ok": ok,
"detail": detail,
"blocked": not ok,
"title": "无法使用 TradingView",
"message": None if ok else TV_CONNECTIVITY_MESSAGE,
"wiki_url": TV_CLOUD_SERVER_WIKI_URL,
}
@app.get("/api/realtime/strategies")
def api_realtime_strategies() -> dict[str, Any]:
"""已保存的 best_*.json 策略,供因子来源下拉。"""
-120
View File
@@ -1,120 +0,0 @@
"""东方财富数据源(A 股 / 指数,直连 push2his HTTP API,无需第三方库)。"""
from __future__ import annotations
from datetime import datetime, timezone
from web.data_sources.base import Bar, DataSource, DataSourceUnavailable
# 项目周期 -> 东财 klt
_KLT = {
"1m": "1",
"5m": "5",
"15m": "15",
"30m": "30",
"1h": "60",
"1d": "101",
"1w": "102",
"1M": "103",
}
_PRESETS = ["600519", "000001", "300750", "601318", "000858", "sh000300", "sz399006"]
_KLINE_URL = "https://push2his.eastmoney.com/api/qt/stock/kline/get"
def _secid(symbol: str) -> str:
"""把 A 股代码转成东财 secid(1.=上交所, 0.=深交所)。"""
s = symbol.strip().lower()
if s.startswith("sh"):
return f"1.{s[2:]}"
if s.startswith("sz"):
return f"0.{s[2:]}"
code = s
if not code.isdigit() or len(code) != 6:
raise DataSourceUnavailable("A 股代码需为 6 位数字,或 sh/sz 前缀的指数代码")
# 6/5/9 开头(含指数 000xxx 需前缀 sh)→ 上交所;0/3 → 深交所
if code[0] in ("6", "5", "9") or code.startswith("11") or code.startswith("13"):
return f"1.{code}"
return f"0.{code}"
def _parse_dt(s: str) -> int:
for fmt in ("%Y-%m-%d %H:%M", "%Y-%m-%d"):
try:
return int(datetime.strptime(s, fmt).replace(tzinfo=timezone.utc).timestamp())
except ValueError:
continue
return 0
class EastMoneySource(DataSource):
kind = "eastmoney"
label = "东方财富"
def available(self) -> tuple[bool, str]:
try:
import requests # noqa: F401
except ImportError:
return (False, "未安装 requests:pip install requests")
return (True, "A 股 / 指数 · 6 位代码或 sh/sz 指数")
def supported_timeframes(self) -> list[str]:
return list(_KLT.keys())
def preset_symbols(self) -> list[str]:
return list(_PRESETS)
def fetch_bars(
self, symbol: str, timeframe: str, n: int, drop_forming: bool = True
) -> list[Bar]:
if timeframe not in _KLT:
raise DataSourceUnavailable(f"东方财富不支持周期 {timeframe}")
try:
import requests
except ImportError as exc:
raise DataSourceUnavailable("未安装 requests") from exc
secid = _secid(symbol)
lmt = n + 2 if drop_forming else n + 1
params = {
"secid": secid,
"klt": _KLT[timeframe],
"fqt": "1", # 前复权
"fields1": "f1,f2,f3,f4,f5",
"fields2": "f51,f52,f53,f54,f55,f56,f57",
"beg": "0",
"end": "20500101",
"lmt": str(lmt),
}
headers = {"User-Agent": "Mozilla/5.0", "Referer": "https://quote.eastmoney.com/"}
try:
resp = requests.get(_KLINE_URL, params=params, headers=headers, timeout=8)
resp.raise_for_status()
payload = resp.json()
except Exception as exc:
raise DataSourceUnavailable(f"东方财富拉取失败: {exc}") from exc
klines = (payload.get("data") or {}).get("klines") or []
if not klines:
raise DataSourceUnavailable(f"东方财富无数据:{symbol}")
bars: list[Bar] = []
for line in klines:
# date,open,close,high,low,volume,amount
parts = line.split(",")
if len(parts) < 6:
continue
bars.append(
Bar(
ts=_parse_dt(parts[0]),
open=float(parts[1]),
close=float(parts[2]),
high=float(parts[3]),
low=float(parts[4]),
volume=float(parts[5]),
)
)
# 已升序;剔除最后一根(盘中为正在形成的 bar)
if drop_forming and len(bars) > 1:
bars = bars[:-1]
return bars[-n:]
+4 -5
View File
@@ -5,11 +5,10 @@ import threading
from web.data_sources.base import DataSource
# 前端下拉可见的数据源(okx / tongdaxin 实现保留,暂不展示)
SOURCE_KINDS: tuple[tuple[str, str], ...] = (
("mt5", "MT5"),
("tradingview", "TradingView"),
("eastmoney", "东方财富"),
("tongdaxin", "通达信"),
)
_INSTANCES: dict[str, DataSource] = {}
@@ -23,9 +22,9 @@ def _build(kind: str) -> DataSource:
if kind == "tradingview":
from web.data_sources.tradingview_source import TradingViewSource
return TradingViewSource()
if kind == "eastmoney":
from web.data_sources.eastmoney_source import EastMoneySource
return EastMoneySource()
if kind == "okx":
from web.data_sources.okx_source import OKXSource
return OKXSource()
if kind == "tongdaxin":
from web.data_sources.tongdaxin_source import TongdaxinSource
return TongdaxinSource()
+142
View File
@@ -0,0 +1,142 @@
"""OKX 数据源(公开 REST,USDT 永续 / 现货)。"""
from __future__ import annotations
import json
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from web.data_sources.base import Bar, DataSource, DataSourceUnavailable
OKX_BASE = "https://www.okx.com"
# 项目周期 -> OKX bar
_TF = {
"1m": "1m",
"5m": "5m",
"15m": "15m",
"30m": "30m",
"1h": "1H",
"4h": "4H",
"1d": "1D",
"1w": "1W",
"1M": "1M",
}
_PRESETS = [
"BTCUSDT",
"ETHUSDT",
"SOLUSDT",
"XRPUSDT",
"DOGEUSDT",
"BNBUSDT",
"ADAUSDT",
"LINKUSDT",
]
def _normalize_inst_id(symbol: str) -> str:
"""BTCUSDT / BTC-USDT / BTC-USDT-SWAP -> BTC-USDT-SWAP(优先永续)。"""
s = (symbol or "").strip().upper().replace("/", "-").replace("_", "-")
if not s:
raise DataSourceUnavailable("请填写品种")
if s.endswith("-SWAP"):
return s
if s.count("-") >= 1:
# BTC-USDT / ETH-USDT-SWAP
parts = s.split("-")
if len(parts) >= 2 and parts[1] in ("USDT", "USD", "USDC"):
return f"{parts[0]}-{parts[1]}-SWAP"
return s
# BTCUSDT / BTCUSD
for quote in ("USDT", "USDC", "USD"):
if s.endswith(quote) and len(s) > len(quote):
base = s[: -len(quote)]
return f"{base}-{quote}-SWAP"
raise DataSourceUnavailable(f"无法识别 OKX 品种:{symbol}(示例 BTCUSDT)")
def _okx_get(path: str, params: dict[str, str], retries: int = 3) -> list:
query = urllib.parse.urlencode(params)
url = f"{OKX_BASE}{path}?{query}"
last_err: Exception | None = None
for attempt in range(retries):
try:
req = urllib.request.Request(url, headers={"User-Agent": "AlphaMaster/1.0"})
with urllib.request.urlopen(req, timeout=20) as resp:
body = json.loads(resp.read().decode("utf-8"))
if body.get("code") != "0":
raise RuntimeError(f"OKX API {body.get('code')}: {body.get('msg')}")
return body.get("data") or []
except Exception as exc: # noqa: BLE001
last_err = exc
if attempt + 1 < retries:
time.sleep(min(8, 2**attempt))
raise DataSourceUnavailable(f"OKX 请求失败: {last_err}")
class OKXSource(DataSource):
kind = "okx"
label = "OKX"
def __init__(self) -> None:
self._lock = threading.Lock()
def available(self) -> tuple[bool, str]:
return (True, "公开行情 · USDT 永续(示例 BTCUSDT)")
def supported_timeframes(self) -> list[str]:
return list(_TF.keys())
def preset_symbols(self) -> list[str]:
return list(_PRESETS)
def fetch_bars(
self, symbol: str, timeframe: str, n: int, drop_forming: bool = True
) -> list[Bar]:
if timeframe not in _TF:
raise DataSourceUnavailable(f"OKX 不支持周期 {timeframe}")
inst_id = _normalize_inst_id(symbol)
bar = _TF[timeframe]
want = min(max(n + 2, 20), 300)
with self._lock:
try:
raw = _okx_get(
"/api/v5/market/candles",
{"instId": inst_id, "bar": bar, "limit": str(want)},
)
except DataSourceUnavailable:
raw = []
if (not raw) and inst_id.endswith("-SWAP"):
spot = inst_id[: -len("-SWAP")]
raw = _okx_get(
"/api/v5/market/candles",
{"instId": spot, "bar": bar, "limit": str(want)},
)
if not raw:
raise DataSourceUnavailable(f"OKX 无数据:{symbol}")
bars: list[Bar] = []
for item in raw:
# [ts_ms, o, h, l, c, vol, volCcy, volCcyQuote, confirm]
bars.append(
Bar(
ts=int(int(item[0]) // 1000),
open=float(item[1]),
high=float(item[2]),
low=float(item[3]),
close=float(item[4]),
volume=float(item[5] or 0.0),
)
)
bars.sort(key=lambda b: b.ts)
if drop_forming and len(bars) > 1:
newest = raw[0] # API 返回降序
# confirm==0 未收盘;缺字段时保守剔除最新一根
if len(newest) <= 8 or str(newest[8]) == "0":
bars = bars[:-1]
return bars[-n:]
@@ -0,0 +1,95 @@
"""Probe whether this machine can reach TradingView via tvdatafeed.
Ported from PA_Agent ``pa_agent.data.tradingview_connectivity``.
"""
from __future__ import annotations
import concurrent.futures
import logging
import time
logger = logging.getLogger(__name__)
_DEFAULT_PROBE_TIMEOUT_S = 20.0
_DEFAULT_PROBE_ATTEMPTS = 3
_DEFAULT_RETRY_DELAY_S = 3.0
# Frontend / watch message sentinel — mirrors PA_Agent VPN dialog trigger
TV_CONNECTIVITY_BLOCKED = "TV_CONNECTIVITY_BLOCKED"
TV_CLOUD_SERVER_WIKI_URL = "https://my.feishu.cn/wiki/FuqnwkPwdiCLhQkPloKc7r1lntg"
TV_CONNECTIVITY_MESSAGE = (
"当前设备无法连接 TradingView 数据服务,将无法获取以下 K 线数据:\n"
" · A 股(上证 SSE、深证 SZSE)\n"
" · 港股(HKEX)\n"
" · 美股及指数(NYSE、NASDAQ、SP)\n"
" · 外汇、贵金属、商品期货\n\n"
"解决方案:\n"
" · 把你的VPN工具设成全局,并开启TUN(虚拟网卡)模式,如果还不行:\n"
" · 使用云服务器部署本程序(推荐)—— 云服务器可正常连接 TradingView\n"
" · 或切换回 MT5 数据源,仅使用 MT5 提供的品种数据"
)
def _probe_once(*, timeout_s: float) -> tuple[bool, str | None, bool]:
"""Single probe. Returns (ok, failure_detail, retryable)."""
def _probe() -> None:
from tvDatafeed import Interval, TvDatafeed # type: ignore[import]
tv = TvDatafeed()
try:
setattr(tv, "_TvDatafeed__ws_timeout", min(10.0, float(timeout_s)))
except Exception:
pass
df = tv.get_hist(
symbol="XAUUSD",
exchange="OANDA",
interval=Interval.in_1_minute,
n_bars=2,
)
if df is None or getattr(df, "empty", True):
raise RuntimeError("TradingView 返回空数据")
try:
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
fut = pool.submit(_probe)
fut.result(timeout=timeout_s)
return True, None, False
except concurrent.futures.TimeoutError:
logger.warning(
"TradingView connectivity probe timed out after %.0fs", timeout_s
)
return False, "连接超时", True
except ImportError as exc:
logger.warning(
"TradingView connectivity probe: tvDatafeed not installed: %s", exc
)
return False, str(exc), False
except Exception as exc: # noqa: BLE001
logger.warning("TradingView connectivity probe failed: %s", exc)
return False, str(exc), True
def check_tradingview_connectivity(
*,
timeout_s: float = _DEFAULT_PROBE_TIMEOUT_S,
max_attempts: int = _DEFAULT_PROBE_ATTEMPTS,
retry_delay_s: float = _DEFAULT_RETRY_DELAY_S,
) -> tuple[bool, str | None]:
"""Try a minimal OANDA:XAUUSD fetch with retries; return (ok, failure_detail)."""
attempts = max(1, int(max_attempts))
last_detail: str | None = None
for attempt in range(1, attempts + 1):
ok, detail, retryable = _probe_once(timeout_s=timeout_s)
if ok:
return True, None
last_detail = detail
if not retryable or attempt >= attempts:
break
time.sleep(max(0.0, float(retry_delay_s)))
return False, last_detail
+25 -1
View File
@@ -160,6 +160,7 @@ class WatchTask:
),
"updated_at": self.updated_at,
"message": self.message,
"tv_blocked": self.message == "TV_CONNECTIVITY_BLOCKED",
"warn": self.warn,
"threshold": min_exposure(),
"history": list(self.history),
@@ -187,6 +188,7 @@ class RealtimeManager:
# K线缓存:(kind,symbol,tf) -> (monotonic_ts, bars)
self._bar_cache: dict[tuple[str, str, str], tuple[float, list]] = {}
self._loaded = False
self._tv_blocked_until = 0.0
# ── 持久化 ──────────────────────────────────────────────────────────
def load_persisted(self) -> None:
@@ -389,7 +391,29 @@ class RealtimeManager:
except Exception:
pass
except Exception as exc: # noqa: BLE001
self._set_error(task, str(exc))
msg = str(exc)
if task.source == "tradingview":
try:
from web.data_sources.tradingview_connectivity import (
TV_CONNECTIVITY_BLOCKED,
check_tradingview_connectivity,
)
now_m = time.monotonic()
if now_m < self._tv_blocked_until:
msg = TV_CONNECTIVITY_BLOCKED
else:
ok, _detail = check_tradingview_connectivity(
timeout_s=12.0, max_attempts=1, retry_delay_s=0.0
)
if not ok:
self._tv_blocked_until = now_m + 120.0
msg = TV_CONNECTIVITY_BLOCKED
else:
self._tv_blocked_until = 0.0
except Exception:
pass
self._set_error(task, msg)
finally:
with self._inflight_lock:
self._inflight.discard(task.id)
+84 -1
View File
@@ -6,6 +6,7 @@ from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parents[1]
SETTINGS_PATH = PROJECT_ROOT / "web_settings.json"
STRATEGIES_DIR = PROJECT_ROOT / "strategies"
_DEFAULT = {
"last_data_file": "",
@@ -35,6 +36,70 @@ def _as_pct(value, default: float) -> float:
return v
def _is_ephemeral_data_path(path: str) -> bool:
"""Stale pytest temp parquet paths (not valid training data)."""
norm = str(path or "").replace("\\", "/").lower()
if "pytest-of-" not in norm:
return False
return (
"/appdata/local/temp/" in norm
or norm.startswith("/tmp/")
or "/temp/" in norm
)
def _is_production_settings_path() -> bool:
try:
return SETTINGS_PATH.resolve() == (PROJECT_ROOT / "web_settings.json").resolve()
except OSError:
return False
def _is_usable_data_file(path: str) -> bool:
p = Path(str(path or "").strip())
return p.is_file() and p.suffix.lower() == ".parquet"
def _should_replace_last_data_file(path: str) -> bool:
cur = str(path or "").strip()
if not cur:
return True
if not Path(cur).is_file():
return True
return _is_ephemeral_data_path(cur)
def _data_file_from_strategy_json(path: str) -> str | None:
p = Path(str(path or "").strip())
if not p.is_file():
return None
try:
data = json.loads(p.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
return None
candidate = str(data.get("data_file") or "").strip()
if _is_usable_data_file(candidate):
return str(Path(candidate).resolve())
return None
def _recover_last_data_file(current: dict) -> str:
cur = str(current.get("last_data_file") or "").strip()
if cur and not _should_replace_last_data_file(cur):
return str(Path(cur).resolve())
for strategy_path in (
str(current.get("last_strategy_file") or "").strip(),
*(str(p) for p in sorted(STRATEGIES_DIR.glob("best_*.json")) if p.is_file()),
):
if not strategy_path:
continue
candidate = _data_file_from_strategy_json(strategy_path)
if candidate:
return candidate
return cur
def load_settings() -> dict:
if not SETTINGS_PATH.exists():
return dict(_DEFAULT)
@@ -75,13 +140,31 @@ def load_settings() -> dict:
out["feishu_enabled"] = bool(out.get("feishu_enabled", False))
out["feishu_webhook_url"] = str(out.get("feishu_webhook_url") or "").strip()
out["feishu_secret"] = str(out.get("feishu_secret") or "").strip()
recovered = _recover_last_data_file(out)
if recovered != out.get("last_data_file") and _is_production_settings_path():
out["last_data_file"] = recovered
if recovered:
SETTINGS_PATH.write_text(
json.dumps(out, indent=2, ensure_ascii=False),
encoding="utf-8",
)
elif recovered != out.get("last_data_file"):
out["last_data_file"] = recovered
return out
def save_settings(data: dict) -> dict:
current = load_settings()
if "last_data_file" in data:
current["last_data_file"] = str(data["last_data_file"] or "").strip()
path = str(data["last_data_file"] or "").strip()
if (
path
and _is_ephemeral_data_path(path)
and _is_production_settings_path()
):
data = {k: v for k, v in data.items() if k != "last_data_file"}
else:
current["last_data_file"] = path
if "last_strategy_file" in data:
current["last_strategy_file"] = str(data["last_strategy_file"] or "").strip()
if "debug_mode" in data:
+143 -9
View File
@@ -662,6 +662,7 @@ async function refreshOverview() {
}
await refreshDebugLogs();
refreshAiProviderStatus();
}
async function loadConfig() {
@@ -710,6 +711,16 @@ function updateBtCostHint() {
hint.textContent = `单边成本 ${fee}%`;
}
async function refreshAiProviderStatus() {
try {
const status = await fetchJSON("/api/ai/providers", { silent: true });
window.__aiProviderStatus = status;
} catch (_) {
/* keep previous snapshot */
}
updateAiChannelHint();
}
async function initAiPanel(cfg) {
const keyInput = $("aiApiKeyInput");
if (!keyInput) return;
@@ -719,13 +730,14 @@ async function initAiPanel(cfg) {
keyInput.value = cfg.ai_provider;
}
try {
const status = await fetchJSON("/api/ai/providers");
window.__aiProviderStatus = status;
} catch (_) {
window.__aiProviderStatus = null;
await refreshAiProviderStatus();
if (!keyInput.dataset.aiStatusBound) {
keyInput.dataset.aiStatusBound = "1";
keyInput.addEventListener("input", () => {
updateAiChannelHint();
refreshAiProviderStatus();
});
}
updateAiChannelHint();
}
function resolveAiFromKey(raw) {
@@ -754,7 +766,9 @@ function updateAiChannelHint() {
if (headHint) headHint.textContent = "DeepSeek · deepseek-v4-flash";
hint.textContent = "当前:DeepSeek(deepseek-v4-flash · https://api.deepseek.com)。";
} else if (resolved.provider === "openclaw") {
if (headHint) headHint.textContent = row?.available ? "openclaw (QClaw) · 已匹配" : "openclaw (QClaw) · 未就绪";
if (headHint) {
headHint.textContent = row?.available ? "openclaw (QClaw) · 已匹配" : "openclaw (QClaw) · 未就绪";
}
hint.textContent = row?.hint || "已匹配 openclaw:将自动使用本地 QClaw token。";
} else {
if (headHint) headHint.textContent = row?.available ? "openclaw_wb · 已匹配" : "openclaw_wb · 未就绪";
@@ -785,6 +799,8 @@ async function runAiAnalyze() {
return;
}
await refreshAiProviderStatus();
if (btn) btn.disabled = true;
view.className = "ai-answer loading";
view.textContent = `正在通过 ${resolved.provider} 连接并流式分析…`;
@@ -1795,6 +1811,19 @@ let rtImportedStrategy = null; // {path, name}
let rtGridSig = "";
let rtServerSkew = 0; // server_time - local_now(秒)
let rtCountdownTimer = null;
let rtTvBlockedShownAt = 0;
let rtTvWikiUrl = "https://my.feishu.cn/wiki/FuqnwkPwdiCLhQkPloKc7r1lntg";
const RT_TV_BLOCKED_MSG =
"当前设备无法连接 TradingView 数据服务,将无法获取以下 K 线数据:\n" +
" · A 股(上证 SSE、深证 SZSE)\n" +
" · 港股(HKEX)\n" +
" · 美股及指数(NYSE、NASDAQ、SP)\n" +
" · 外汇、贵金属、商品期货\n\n" +
"解决方案:\n" +
" · 把你的VPN工具设成全局,并开启TUN(虚拟网卡)模式,如果还不行:\n" +
" · 使用云服务器部署本程序(推荐)—— 云服务器可正常连接 TradingView\n" +
" · 或切换回 MT5 数据源,仅使用 MT5 提供的品种数据";
const RT_TV_BLOCKED_CODE = "TV_CONNECTIVITY_BLOCKED";
const RT_DIR = {
LONG: { label: "↑ 预期上涨", cls: "rt-long", color: "#4ade80" },
@@ -2124,6 +2153,10 @@ async function rtAddWatch() {
const btn = $("rtAddBtn");
if (btn) btn.disabled = true;
try {
if (source === "tradingview") {
const reachable = await rtEnsureTradingViewReachable();
if (!reachable) return;
}
await fetchJSON("/api/realtime/watch", {
method: "POST",
headers: { "Content-Type": "application/json" },
@@ -2140,6 +2173,97 @@ async function rtAddWatch() {
}
}
async function rtEnsureTradingViewReachable() {
const picked = $("rtStrategyPicked");
if (picked) {
picked.textContent = "正在检测 TradingView 连通性…";
picked.classList.remove("bad");
}
try {
const res = await fetchJSON("/api/realtime/tradingview/probe", {
method: "POST",
silent: true,
});
if (res && res.ok) {
if (picked) picked.textContent = "";
return true;
}
if (res?.wiki_url) rtTvWikiUrl = res.wiki_url;
await showTvBlockedDialog(res?.message || RT_TV_BLOCKED_MSG);
if (picked) {
picked.textContent = "TradingView 不可用,请开 VPN 或换云服务器";
picked.classList.add("bad");
}
return false;
} catch (e) {
await showTvBlockedDialog(RT_TV_BLOCKED_MSG);
if (picked) {
picked.textContent = "TradingView 检测失败: " + (e.message || e);
picked.classList.add("bad");
}
return false;
}
}
function showTvBlockedDialog(message) {
const now = Date.now();
// 避免轮询反复弹出
if (now - rtTvBlockedShownAt < 60_000) {
return Promise.resolve("dedupe");
}
rtTvBlockedShownAt = now;
const modal = $("tvBlockedModal");
const body = $("tvBlockedBody");
if (!modal || !body) {
window.alert(message || RT_TV_BLOCKED_MSG);
return Promise.resolve("alert");
}
body.textContent = message || RT_TV_BLOCKED_MSG;
modal.hidden = false;
return new Promise((resolve) => {
modal._tvResolve = resolve;
});
}
function closeTvBlockedDialog(choice) {
const modal = $("tvBlockedModal");
if (modal) modal.hidden = true;
const resolve = modal && modal._tvResolve;
if (resolve) {
modal._tvResolve = null;
resolve(choice || "cancel");
}
}
async function onTvBlockedSwitchMt5() {
const sel = $("rtSourceSelect");
if (sel) {
const hasMt5 = [...sel.options].some((o) => o.value === "mt5");
if (hasMt5) {
sel.value = "mt5";
onRtSourceChange();
}
}
closeTvBlockedDialog("mt5");
}
function onTvBlockedOpenCloud() {
try {
window.open(rtTvWikiUrl, "_blank", "noopener,noreferrer");
} catch (_) {}
closeTvBlockedDialog("cloud");
}
function maybeShowTvBlockedFromWatches(watches) {
const hit = (watches || []).find(
(w) =>
w.source === "tradingview" &&
(w.tv_blocked || w.message === RT_TV_BLOCKED_CODE)
);
if (!hit) return;
showTvBlockedDialog(RT_TV_BLOCKED_MSG);
}
async function rtRemoveWatch(id) {
try {
await fetchJSON("/api/realtime/unwatch", {
@@ -2200,6 +2324,7 @@ async function refreshRealtime() {
}
}
renderRealtimeGrid(st.watches || []);
maybeShowTvBlockedFromWatches(st.watches || []);
ensureRtCountdownTimer();
tickRtCountdowns();
}
@@ -2275,9 +2400,13 @@ function renderRealtimeGrid(watches) {
const srcLabel = (rtSourceById[w.source] || {}).label || w.source;
const factorText = w.factor_value != null ? Number(w.factor_value).toFixed(4) : "—";
const warn = w.warn ? `<div class="rt-warn" title="${escHtml(w.warn)}">⚠ ${escHtml(w.warn)}</div>` : "";
const displayMsg =
w.message === RT_TV_BLOCKED_CODE || w.tv_blocked
? "无法连接 TradingView:请开启全局 VPN(TUN)或使用云服务器"
: w.message;
const msg =
w.state !== "ok" && w.message
? `<div class="rt-msg">${escHtml(w.message)}</div>`
w.state !== "ok" && displayMsg
? `<div class="rt-msg">${escHtml(displayMsg)}</div>`
: "";
const sizeText = plain ? plain.size : "—";
return `
@@ -2387,6 +2516,11 @@ async function init() {
if ($("rtStrategySelect")) $("rtStrategySelect").addEventListener("change", onRtStrategyChange);
if ($("rtBrowseStrategyBtn")) $("rtBrowseStrategyBtn").addEventListener("click", rtBrowseStrategy);
if ($("rtAddBtn")) $("rtAddBtn").addEventListener("click", rtAddWatch);
if ($("tvBlockedMt5Btn")) $("tvBlockedMt5Btn").addEventListener("click", onTvBlockedSwitchMt5);
if ($("tvBlockedCloudBtn")) $("tvBlockedCloudBtn").addEventListener("click", onTvBlockedOpenCloud);
document.querySelectorAll("[data-close-tv-blocked]").forEach((el) => {
el.addEventListener("click", () => closeTvBlockedDialog("cancel"));
});
if ($("rtFeishuSaveBtn")) $("rtFeishuSaveBtn").addEventListener("click", saveRtFeishuSettings);
if ($("rtFeishuTestBtn")) $("rtFeishuTestBtn").addEventListener("click", testRtFeishu);
if ($("rtFeishuHelpBtn")) $("rtFeishuHelpBtn").addEventListener("click", openRtFeishuHelpModal);
+13 -1
View File
@@ -540,6 +540,18 @@
</div>
</div>
<div class="modal" id="tvBlockedModal" hidden>
<div class="modal-backdrop" data-close-tv-blocked></div>
<div class="modal-card" role="dialog" aria-labelledby="tvBlockedTitle">
<h3 id="tvBlockedTitle">无法使用 TradingView</h3>
<pre class="modal-body modal-tv-blocked-body" id="tvBlockedBody"></pre>
<div class="modal-actions">
<button type="button" class="btn btn-secondary" id="tvBlockedMt5Btn">切换回 MT5</button>
<button type="button" class="btn btn-primary" id="tvBlockedCloudBtn">使用云服务器</button>
</div>
</div>
</div>
<div class="modal" id="errorModal" hidden>
<div class="modal-backdrop" data-close-error></div>
<div class="modal-card modal-card-error" role="dialog" aria-labelledby="errorModalTitle">
@@ -554,6 +566,6 @@
</div>
<script src="/static/bg.js?v=1"></script>
<script src="/static/app.js?v=58"></script>
<script src="/static/app.js?v=59"></script>
</body>
</html>
+13
View File
@@ -1179,6 +1179,19 @@ td:nth-child(2) { color: var(--accent); font-family: var(--mono); }
word-break: break-word;
}
.modal-tv-blocked-body {
margin: 0 0 18px;
padding: 0;
color: var(--text);
font-family: inherit;
font-size: 0.92rem;
line-height: 1.65;
white-space: pre-wrap;
word-break: break-word;
background: transparent;
border: none;
}
.data-file-card .path.data-file-missing {
color: #fca5a5;
}
+77 -3
View File
@@ -6,6 +6,7 @@ import re
from pathlib import Path
from typing import Any
from data_pipeline.parquet_manager import inspect_parquet_file
from model_core.vocab import VOCAB_VERSION
from web.progress import (
STRATEGIES_DIR,
@@ -14,6 +15,7 @@ from web.progress import (
_load_strategy,
checkpoint_glob,
)
from web.settings import load_settings
_BEST_NAME_RE = re.compile(r"^best_(.+)\.json$", re.IGNORECASE)
_STRATEGY_EXPORT_RE = re.compile(
@@ -37,7 +39,66 @@ def symbol_from_strategy_path(path: Path) -> str | None:
return None
def inspect_strategy_file(path: str) -> dict[str, Any]:
def _resolve_data_file_for_symbol(
symbol: str,
data_file: str | None,
*,
data_file_hint: str | None = None,
) -> tuple[str | None, str | None]:
"""Fill missing strategy data_file from training settings / job hint."""
sym = (symbol or "").strip()
if data_file:
p = Path(str(data_file))
if p.exists():
try:
info = inspect_parquet_file(str(p.resolve()))
if not sym or info.get("symbol") == sym:
return str(p.resolve()), info.get("timeframe")
except Exception:
return str(p.resolve()), None
for candidate in (data_file_hint, load_settings().get("last_data_file") or ""):
path = str(candidate or "").strip()
if not path:
continue
p = Path(path)
if not p.exists():
continue
try:
info = inspect_parquet_file(str(p.resolve()))
except Exception:
continue
if sym and info.get("symbol") != sym:
continue
return str(p.resolve()), info.get("timeframe")
return data_file, None
def _apply_data_file_fallback(
payload: dict[str, Any],
*,
data_file_hint: str | None = None,
) -> dict[str, Any]:
symbol = str(payload.get("symbol") or "").strip()
data_file, timeframe = _resolve_data_file_for_symbol(
symbol,
payload.get("data_file"),
data_file_hint=data_file_hint,
)
if data_file:
payload["data_file"] = data_file
if timeframe and not payload.get("timeframe"):
payload["timeframe"] = timeframe
if data_file and not payload.get("mode"):
payload["mode"] = "parquet_file"
return payload
def inspect_strategy_file(
path: str,
*,
data_file_hint: str | None = None,
) -> dict[str, Any]:
p = Path(path)
if not p.exists():
raise FileNotFoundError(f"文件不存在: {p}")
@@ -73,6 +134,14 @@ def inspect_strategy_file(path: str) -> dict[str, Any]:
data_file = data.get("data_file")
mode = data.get("mode")
data_file, tf_fallback = _resolve_data_file_for_symbol(
symbol or "",
data_file,
data_file_hint=data_file_hint,
)
if tf_fallback and not timeframe:
timeframe = tf_fallback
data_file_exists = bool(data_file) and Path(str(data_file)).exists()
return {
@@ -119,7 +188,11 @@ def _step_from_export_name(path: Path) -> int:
return 0
def sync_best_strategy_for_symbol(symbol: str) -> dict[str, Any] | None:
def sync_best_strategy_for_symbol(
symbol: str,
*,
data_file_hint: str | None = None,
) -> dict[str, Any] | None:
"""在策略文件与检查点中选出最高分策略,写入 strategies/best_{symbol}.json。"""
candidates: list[tuple[float, list[int], int]] = []
@@ -169,5 +242,6 @@ def sync_best_strategy_for_symbol(symbol: str) -> dict[str, Any] | None:
for key in ("timeframe", "data_file", "mode", "train_steps"):
if strat.get(key) is not None:
payload[key] = strat[key]
payload = _apply_data_file_fallback(payload, data_file_hint=data_file_hint)
out_path.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8")
return inspect_strategy_file(str(out_path.resolve()))
return inspect_strategy_file(str(out_path.resolve()), data_file_hint=data_file_hint)