Merge pull request #45 from TNT-Likely/fix/proxy-and-sqlite-lock

TNT-Likely · web-flow · commit eebd90257abd · 2026-06-18T10:34:43.000+08:00
fix: 生产代理拦截行情接口 + 缓解 SQLite database is locked
diff --git a/src/collectors/akshare_collector.py b/src/collectors/akshare_collector.py
@@ -130,7 +130,7 @@ def _fetch_tencent_quotes(symbols: list[str]) -> list[dict]:
     if not symbols:
         return []
     url = TENCENT_QUOTE_URL + ",".join(symbols)
-    with httpx.Client() as client:
+    with httpx.Client(trust_env=False) as client:  # 行情直连,绕过 env 代理
         resp = client.get(url, timeout=10)
         content = resp.content.decode("gbk", errors="ignore")
 
diff --git a/src/collectors/capital_flow_collector.py b/src/collectors/capital_flow_collector.py
@@ -77,7 +77,9 @@ def get_capital_flow(self, symbol: str) -> CapitalFlow | None:
         }
 
         try:
-            with httpx.Client(follow_redirects=True, timeout=8) as client:
+            with httpx.Client(
+                follow_redirects=True, timeout=8, trust_env=False
+            ) as client:  # 行情直连,绕过 env 代理(生产代理会拦 push2his.eastmoney)
                 resp = client.get(EASTMONEY_FLOW_URL, params=params, headers=headers)
                 data = resp.json()
 
diff --git a/src/collectors/events_collector.py b/src/collectors/events_collector.py
@@ -116,7 +116,7 @@ async def fetch_events(
                     verify=self.verify_ssl,
                     headers=headers,
                     follow_redirects=True,
-                    trust_env=True,
+                    trust_env=False,  # 不吃 env 代理(Telegram/AI 用),仅用显式配置的 self.proxy
                     proxy=self.proxy,
                 ) as client:
                     resp = await client.get(self.API_URL, params=params)
diff --git a/src/collectors/kline_collector.py b/src/collectors/kline_collector.py
@@ -52,7 +52,10 @@ def _fetch_stooq_us_klines(symbol: str) -> list[KlineData]:
         try:
             timeout = 12 + attempt * 6
             with httpx.Client(
-                follow_redirects=True, timeout=timeout, headers=headers
+                follow_redirects=True,
+                timeout=timeout,
+                headers=headers,
+                trust_env=False,  # 行情直连,绕过 env 代理(生产代理会拦行情接口)
             ) as client:
                 resp = client.get(url, params=params)
                 resp.raise_for_status()
@@ -155,6 +158,7 @@ def _fetch_eastmoney_klines(
                 follow_redirects=True,
                 timeout=12 + attempt * 6,
                 headers=headers,
+                trust_env=False,  # 行情直连,绕过 env 代理(生产代理会拦 push2his.eastmoney)
             ) as client:
                 resp = client.get(EASTMONEY_KLINE_URL, params=params)
                 resp.raise_for_status()
@@ -492,7 +496,9 @@ def get_klines(self, symbol: str, days: int = 60) -> list[KlineData]:
         }
 
         try:
-            with httpx.Client(follow_redirects=True, timeout=10) as client:
+            with httpx.Client(
+                follow_redirects=True, timeout=10, trust_env=False
+            ) as client:  # 行情直连,绕过 env 代理(生产代理会拦行情接口)
                 resp = client.get(TENCENT_KLINE_URL, params=params)
                 text = resp.text
 
@@ -550,7 +556,9 @@ def get_klines(self, symbol: str, days: int = 60) -> list[KlineData]:
 
             # CN/HK: Tencent 在高 days 时可能只返回近几年，尝试 Eastmoney 补全更长历史
             if self.market in (MarketCode.CN, MarketCode.HK):
-                need_em = days >= 500 or len(klines) < max(120, int(days * 0.6))
+                # 仅当腾讯返回不足时才回退东财。去掉原 `days >= 500` 的无条件触发——
+                # 否则评估循环(days=600)每只标的都额外打一次东财,把它打到自我限流。
+                need_em = len(klines) < max(120, int(days * 0.6))
                 if need_em:
                     # 额外放大窗口，提升拿到更长历史的概率
                     em_target_days = min(max(days, 3000), 20000)
diff --git a/src/collectors/news_collector.py b/src/collectors/news_collector.py
@@ -109,7 +109,7 @@ async def fetch_news(self, symbols: list[str] | None = None, since: datetime | N
         if self.cookies:
             headers["Cookie"] = self.cookies
 
-        async with httpx.AsyncClient(timeout=8, headers=headers) as client:
+        async with httpx.AsyncClient(timeout=8, headers=headers, trust_env=False) as client:  # CN 源直连,绕过 env 代理
             tasks = [self._fetch_for_symbol(client, symbol, since) for symbol in a_share_symbols]
             results = await asyncio.gather(*tasks, return_exceptions=True)
 
@@ -284,7 +284,7 @@ async def fetch_with_limit(client, symbol, stock_name):
             "Referer": "https://so.eastmoney.com/",
             "Accept": "*/*",
         }
-        async with httpx.AsyncClient(timeout=8, verify=False, headers=headers) as client:
+        async with httpx.AsyncClient(timeout=8, verify=False, headers=headers, trust_env=False) as client:  # CN 源直连,绕过 env 代理
             tasks = [
                 fetch_with_limit(client, symbol, symbol_names.get(symbol, symbol))
                 for symbol in symbols
@@ -323,7 +323,7 @@ async def fetch_by_keyword(self, keyword: str) -> list[NewsItem]:
             "Referer": "https://so.eastmoney.com/",
             "Accept": "*/*",
         }
-        async with httpx.AsyncClient(timeout=8, verify=False, headers=headers) as client:
+        async with httpx.AsyncClient(timeout=8, verify=False, headers=headers, trust_env=False) as client:  # CN 源直连,绕过 env 代理
             return await self._fetch_for_symbol(client, keyword, keyword, None)
 
     async def _fetch_for_symbol(self, client: httpx.AsyncClient, symbol: str, stock_name: str, since: datetime | None) -> list[NewsItem]:
@@ -482,7 +482,7 @@ async def fetch_news(self, symbols: list[str] | None = None, since: datetime | N
         }
 
         try:
-            async with httpx.AsyncClient(timeout=5, verify=False) as client:
+            async with httpx.AsyncClient(timeout=5, verify=False, trust_env=False) as client:  # CN 源直连,绕过 env 代理
                 resp = await client.get(self.API_URL, params=params)
                 resp.raise_for_status()
                 data = resp.json()
diff --git a/src/core/entry_candidates.py b/src/core/entry_candidates.py
@@ -1709,6 +1709,7 @@ def evaluate_entry_candidate_outcomes(
 
         today = date.today()
         kline_cache: dict[tuple[str, str], list] = {}
+        pending = 0  # 分批提交计数,缩短写事务窗口
 
         for c in candidates:
             snap_day = _parse_day(c.snapshot_date)
@@ -1804,6 +1805,12 @@ def evaluate_entry_candidate_outcomes(
                 db.add(row)
                 stats["evaluated"] += 1
                 existing.add((c.id, horizon))
+                pending += 1
+
+            # 分批提交:累计到阈值即落盘,缩短写事务,避免与 60s 调度器并发写长时间持锁
+            if pending >= 50:
+                db.commit()
+                pending = 0
 
         db.commit()
         return stats
diff --git a/src/core/paper_trading_engine.py b/src/core/paper_trading_engine.py
@@ -470,16 +470,20 @@ def _check_exits(
 
             # 检查信号反转
             if pos.signal_run_id:
-                latest = (
-                    db.query(StrategySignalRun)
-                    .filter(
-                        StrategySignalRun.stock_symbol == pos.stock_symbol,
-                        StrategySignalRun.stock_market == pos.stock_market,
-                        StrategySignalRun.status == "active",
+                # no_autoflush: 信号查询是只读的,不要把本轮累积的持仓现价更新提前 flush——
+                # 否则扫描中途会反复抢 SQLite 写锁,与其它调度器并发写时触发 "database is locked"。
+                # 所有写入统一在本方法末尾 db.commit() 时一次性落盘。
+                with db.no_autoflush:
+                    latest = (
+                        db.query(StrategySignalRun)
+                        .filter(
+                            StrategySignalRun.stock_symbol == pos.stock_symbol,
+                            StrategySignalRun.stock_market == pos.stock_market,
+                            StrategySignalRun.status == "active",
+                        )
+                        .order_by(StrategySignalRun.created_at.desc())
+                        .first()
                     )
-                    .order_by(StrategySignalRun.created_at.desc())
-                    .first()
-                )
                 if latest and latest.action in ("sell", "reduce"):
                     trade = self._close_position(db, account, pos, current_price, "signal_reversal")
                     exit_events.append((pos, trade))
diff --git a/src/core/paper_trading_scheduler.py b/src/core/paper_trading_scheduler.py
@@ -62,6 +62,7 @@ def start(self):
             self._scan_job,
             "interval",
             seconds=self.interval_seconds,
+            jitter=20,  # 抖动错峰,避免与价格提醒扫描每 60s 同刻并发写 SQLite
             id="paper_trading_scan",
             replace_existing=True,
             coalesce=True,
diff --git a/src/core/price_alert_scheduler.py b/src/core/price_alert_scheduler.py
@@ -50,6 +50,7 @@ def start(self):
             self._scan_job,
             "interval",
             seconds=self.interval_seconds,
+            jitter=20,  # 抖动错峰,避免与模拟盘扫描每 60s 同刻并发写 SQLite
             id="price_alert_scan",
             replace_existing=True,
             coalesce=True,
diff --git a/src/core/strategy_engine.py b/src/core/strategy_engine.py
@@ -1567,6 +1567,7 @@ def evaluate_strategy_outcomes(
 
         today = date.today()
         kline_cache: dict[tuple[str, str], list] = {}
+        pending = 0  # 分批提交计数,缩短写事务窗口
 
         for s in signals:
             snap_day = _parse_day(s.snapshot_date)
@@ -1662,6 +1663,12 @@ def evaluate_strategy_outcomes(
                 )
                 stats["evaluated"] += 1
                 existing.add((s.id, horizon))
+                pending += 1
+
+            # 分批提交:累计到阈值即落盘,缩短写事务,避免与 60s 调度器并发写长时间持锁
+            if pending >= 50:
+                db.commit()
+                pending = 0
 
         db.commit()
         return stats
diff --git a/tests/test_kline_collector_proxy.py b/tests/test_kline_collector_proxy.py
@@ -0,0 +1,40 @@
+"""市场数据采集器必须绕过环境代理(trust_env=False)。
+
+生产环境为 Telegram/AI 配置了 HTTP_PROXY/HTTPS_PROXY(LAN Clash 代理),
+但该代理无法服务行情接口(腾讯 gtimg / 东方财富 push2his),会回 "Server
+disconnected without sending a response."。行情采集应直连,不走 env 代理。
+"""
+
+from __future__ import annotations
+
+from src.collectors import kline_collector
+from src.models.market import MarketCode
+
+
+def test_kline_collector_market_data_bypasses_env_proxy(monkeypatch):
+    """行情采集创建的所有 httpx.Client 必须 trust_env=False(直连、不吃 env 代理)。"""
+    captured: list[dict] = []
+
+    class _FakeClient:
+        def __init__(self, **kwargs):
+            captured.append(kwargs)
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *exc):
+            return False
+
+        def get(self, *args, **kwargs):
+            raise RuntimeError("network disabled in test")
+
+    # 屏蔽真实网络与退避 sleep,保持单测快速、离线
+    monkeypatch.setattr(kline_collector.httpx, "Client", _FakeClient)
+    monkeypatch.setattr(kline_collector.time, "sleep", lambda *_: None)
+
+    # days>=500 会同时触发 腾讯主路径 + 东方财富长历史回退,覆盖多个 client
+    kline_collector.KlineCollector(MarketCode.CN).get_klines("000001", days=600)
+
+    assert captured, "应至少创建一个 httpx.Client"
+    for kw in captured:
+        assert kw.get("trust_env") is False, f"行情 client 必须 trust_env=False,实际: {kw}"

Original file line number	Diff line number	Diff line change
`@@ -77,7 +77,9 @@ def get_capital_flow(self, symbol: str) -> CapitalFlow \| None:`
`77`	`77`	`}`
`78`	`78`
`79`	`79`	`try:`
`80`		`- with httpx.Client(follow_redirects=True, timeout=8) as client:`
	`80`	`+ with httpx.Client(`
	`81`	`+ follow_redirects=True, timeout=8, trust_env=False`
	`82`	`+ ) as client: # 行情直连,绕过 env 代理(生产代理会拦 push2his.eastmoney)`
`81`	`83`	`resp = client.get(EASTMONEY_FLOW_URL, params=params, headers=headers)`
`82`	`84`	`data = resp.json()`
`83`	`85`