v4.0.6

Signed-off-by: TIANHE <TIANHE@GMAIL.COM>

Author: tianhe

backend_api_python/app/data_sources/asia_stock_kline.py

@@ -16,9 +16,10 @@
 from __future__ import annotations
 
 import os
+import threading
 import time
 from contextlib import contextmanager
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import Any, Dict, Generator, List, Optional
 
 import pandas as pd
@@ -148,6 +149,41 @@ def _get_twelve_data_api_key() -> str:
     "1W": "1week",
 }
 
+_TD_DAILY_LIMIT_LOCK = threading.Lock()
+_TD_DAILY_LIMIT_UNTIL = 0.0
+
+
+def _twelvedata_daily_limit_active() -> bool:
+    with _TD_DAILY_LIMIT_LOCK:
+        return time.time() < _TD_DAILY_LIMIT_UNTIL
+
+
+def _mark_twelvedata_daily_limited(symbol: str, exchange: str, message: str) -> None:
+    global _TD_DAILY_LIMIT_UNTIL
+    now = datetime.now(timezone.utc)
+    tomorrow = (now + timedelta(days=1)).date()
+    until = datetime(
+        tomorrow.year,
+        tomorrow.month,
+        tomorrow.day,
+        0,
+        5,
+        tzinfo=timezone.utc,
+    ).timestamp()
+    first = False
+    with _TD_DAILY_LIMIT_LOCK:
+        if time.time() >= _TD_DAILY_LIMIT_UNTIL:
+            first = True
+        _TD_DAILY_LIMIT_UNTIL = max(_TD_DAILY_LIMIT_UNTIL, until)
+    if first:
+        logger.warning(
+            "TwelveData daily API credits exhausted for %s/%s; "
+            "skipping TwelveData requests until next UTC day. Last message: %s",
+            symbol,
+            exchange,
+            message,
+        )
+
 
 def _td_symbol_and_exchange(tencent_code: str, is_hk: bool) -> tuple[str, str]:
     """Convert Tencent code to Twelve Data (symbol, exchange).
@@ -183,6 +219,8 @@ def fetch_twelvedata_klines(
     interval = _TD_INTERVAL_MAP.get(timeframe)
     if not interval:
         return []
+    if _twelvedata_daily_limit_active():
+        return []
 
     symbol, exchange = _td_symbol_and_exchange(tencent_code, is_hk)
     merge_factor = _MERGE_FACTOR_MAP.get(timeframe, 1)
@@ -223,7 +261,10 @@ def fetch_twelvedata_klines(
     if data.get("status") != "ok" or "values" not in data:
         code = data.get("code", "")
         msg = data.get("message", str(data))
-        if code == 429 or "API credits" in msg or "minute limit" in msg:
+        msg_l = str(msg).lower()
+        if "api credits" in msg_l or "for the day" in msg_l:
+            _mark_twelvedata_daily_limited(symbol, exchange, msg)
+        elif code == 429 or "minute limit" in msg_l:
             logger.warning("TwelveData rate limit for %s/%s: %s", symbol, exchange, msg)
         elif "Pro" in msg or "Venture" in msg or "upgrading" in msg:
             logger.debug("TwelveData plan limit %s/%s tf=%s: %s", symbol, exchange, timeframe, msg)

backend_api_python/app/data_sources/factory.py

@@ -2,6 +2,9 @@
 数据源工厂
 根据市场类型返回对应的数据源
 """
+import os
+import threading
+import time
 from typing import Dict, List, Any, Optional
 
 from app.data_sources.base import BaseDataSource
@@ -10,6 +13,14 @@
 
 logger = get_logger(__name__)
 
+
+def _env_positive_int(key: str, default: int) -> int:
+    try:
+        value = int(os.getenv(key, str(default)))
+        return value if value > 0 else default
+    except Exception:
+        return default
+
 _MARKET_ALIASES: Dict[str, str] = {
     "crypto": "Crypto",
     "cryptocurrency": "Crypto",
@@ -26,7 +37,22 @@
     "alpaca": "USStock",
     "ibkr": "USStock",
     "cnstock": "CNStock",
+    "cn_stock": "CNStock",
+    "ashare": "CNStock",
+    "a_share": "CNStock",
+    "astock": "CNStock",
+    "a_stock": "CNStock",
+    "cn": "CNStock",
+    "china": "CNStock",
+    "chinastock": "CNStock",
     "hkstock": "HKStock",
+    "hk_stock": "HKStock",
+    "hshare": "HKStock",
+    "h_share": "HKStock",
+    "hkshare": "HKStock",
+    "hk_share": "HKStock",
+    "hk": "HKStock",
+    "hongkong": "HKStock",
     "futures": "Futures",
     "moex": "MOEX",
     "rustock": "MOEX",
@@ -43,10 +69,29 @@ class DataSourceFactory:
     """
 
     _sources: Dict[str, BaseDataSource] = {}
+    _noise_lock = threading.Lock()
+    _noise_seen: Dict[str, tuple[float, int]] = {}
+    _noise_interval_sec = _env_positive_int("LOG_DEDUPE_INTERVAL_SEC", 60)
 
     # Markets that pass through normalize_market unchanged.
     _CANONICAL_MARKETS = ("Crypto", "Forex", "Futures", "USStock", "CNStock", "HKStock", "MOEX")
 
+    @classmethod
+    def _log_limited(cls, level: str, key: str, message: str, *args: Any) -> None:
+        """Log noisy market-data failures at most once per key per interval."""
+        now = time.monotonic()
+        with cls._noise_lock:
+            last, suppressed = cls._noise_seen.get(key, (0.0, 0))
+            if last > 0 and now - last < cls._noise_interval_sec:
+                cls._noise_seen[key] = (last, suppressed + 1)
+                return
+            cls._noise_seen[key] = (now, 0)
+
+        if suppressed:
+            message = f"{message} (suppressed {suppressed} duplicate log(s))"
+        log_fn = getattr(logger, level, logger.warning)
+        log_fn(message, *args)
+
     @classmethod
     def normalize_market(cls, market: str) -> str:
         """
@@ -73,6 +118,14 @@ def normalize_market(cls, market: str) -> str:
         key = raw.lower().replace(" ", "").replace("-", "_")
         if key in _MARKET_ALIASES:
             return _MARKET_ALIASES[key]
+        cls._log_limited(
+            "warning",
+            f"unknown-market:{raw}",
+            "DataSourceFactory.normalize_market(): unknown market %r; "
+            "passing through as-is; downstream get_source() will likely fail.",
+            raw,
+        )
+        return raw
         logger.warning(
             "DataSourceFactory.normalize_market(): unknown market %r — "
             "passing through as-is; downstream get_source() will likely fail.",
@@ -178,16 +231,24 @@ def get_kline(
         Returns:
             K线数据列表
         """
+        m = cls.normalize_market(market or "")
         try:
-            m = cls.normalize_market(market or "")
             source = cls._resolve_source(m, exchange_id=exchange_id, market_type=market_type)
             klines = source.get_kline(symbol, timeframe, limit, before_time, after_time)
 
             klines.sort(key=lambda x: x['time'])
 
             return klines
         except Exception as e:
-            logger.error(f"Failed to fetch K-lines {market}:{symbol} (normalized={cls.normalize_market(market or '')}) - {str(e)}")
+            cls._log_limited(
+                "error",
+                f"kline:{m}:{symbol}:{type(e).__name__}:{str(e)[:160]}",
+                "Failed to fetch K-lines %s:%s (normalized=%s) - %s",
+                market,
+                symbol,
+                m,
+                str(e),
+            )
             return []
 
     @classmethod
@@ -228,14 +289,26 @@ def get_ticker(cls, market: str, symbol: str, exchange_id: Optional[str] = None,
                 ...
             }
         """
+        m = cls.normalize_market(market or "")
         try:
-            m = cls.normalize_market(market or "")
             source = cls._resolve_source(m, exchange_id=exchange_id, market_type=market_type)
             return source.get_ticker(symbol)
         except NotImplementedError:
-            logger.warning(f"get_ticker not implemented for market: {market}")
+            cls._log_limited(
+                "warning",
+                f"ticker-not-implemented:{m}",
+                "get_ticker not implemented for market: %s",
+                market,
+            )
             return {'last': 0, 'symbol': symbol}
         except Exception as e:
-            logger.error(f"Failed to fetch ticker {market}:{symbol} - {str(e)}")
+            cls._log_limited(
+                "error",
+                f"ticker:{m}:{symbol}:{type(e).__name__}:{str(e)[:160]}",
+                "Failed to fetch ticker %s:%s - %s",
+                market,
+                symbol,
+                str(e),
+            )
             return {'last': 0, 'symbol': symbol}
 

backend_api_python/app/services/mfa_service.py

@@ -148,6 +148,7 @@ def start_setup(self, user_id: int, label: str) -> Dict[str, Any]:
                     last_used_counter = 0,
                     confirmed_at = NULL,
                     updated_at = NOW()
+                RETURNING user_id
                 """,
                 (user_id, encrypted),
             )

backend_api_python/app/services/trading_executor.py

@@ -77,6 +77,12 @@ def __init__(self):
         self.kline_service = KlineService()
         # Throttle writes to qd_strategy_logs (heartbeat), per strategy_id -> monotonic time
         self._strategy_ui_log_last_tick_ts = {}  # type: Dict[int, float]
+        self._console_tick_last_ts: Dict[int, float] = {}
+        self._console_tick_lock = threading.Lock()
+        try:
+            self._console_tick_interval_sec = max(1, int(os.getenv("STRATEGY_TICK_LOG_INTERVAL_SEC", "60")))
+        except Exception:
+            self._console_tick_interval_sec = 60
 
         self.max_threads = int(os.getenv('STRATEGY_MAX_THREADS', '64'))
         self._last_start_failure: str = ""
@@ -268,7 +274,18 @@ def _console_print(self, msg: str) -> None:
         Local-only observability: print to stdout so user can see strategy status in console.
         """
         try:
-            print(str(msg or ""), flush=True)
+            text = str(msg or "")
+            if "] tick price=" in text:
+                m = re.match(r"\[strategy:(\d+)\]\s+tick\b", text)
+                if m:
+                    sid = int(m.group(1))
+                    now = time.monotonic()
+                    with self._console_tick_lock:
+                        last = self._console_tick_last_ts.get(sid, 0.0)
+                        if last > 0 and now - last < self._console_tick_interval_sec:
+                            return
+                        self._console_tick_last_ts[sid] = now
+            print(text, flush=True)
         except Exception:
             pass
 
@@ -2527,7 +2544,7 @@ def _is_fatal_error(err: Exception, msg: str) -> bool:
                             if signal_time == 0 or (current_ts - signal_time) < expiration_threshold:
                                 valid_signals.append(s)
                             else:
-                                logger.warning(f"Signal expired and removed: {s}")
+                                logger.debug(f"Signal expired and removed: {s}")
                         if len(valid_signals) != len(pending_signals):
                             pending_signals = valid_signals
 

backend_api_python/app/utils/strategy_runtime_logs.py

@@ -24,9 +24,12 @@ def append_strategy_log(strategy_id: int, level: str, message: str) -> None:
             cur.execute(
                 """
                 INSERT INTO qd_strategy_logs (strategy_id, level, message, timestamp)
-                VALUES (?, ?, ?, ?)
+                SELECT ?, ?, ?, ?
+                WHERE EXISTS (
+                    SELECT 1 FROM qd_strategies_trading WHERE id = ?
+                )
                 """,
-                (sid, lv, msg, datetime.now(timezone.utc)),
+                (sid, lv, msg, datetime.now(timezone.utc), sid),
             )
             db.commit()
             cur.close()

backend_api_python/tests/test_db_log_noise_fixes.py

@@ -0,0 +1,85 @@
+from contextlib import contextmanager
+
+from app.services import mfa_service
+from app.utils import strategy_runtime_logs
+
+
+class _CaptureCursor:
+    def __init__(self):
+        self.calls = []
+        self.closed = False
+
+    def execute(self, sql, params=None):
+        self.calls.append((sql, params))
+
+    def close(self):
+        self.closed = True
+
+
+class _CaptureConn:
+    def __init__(self):
+        self.cursor_obj = _CaptureCursor()
+        self.committed = False
+
+    def cursor(self):
+        return self.cursor_obj
+
+    def commit(self):
+        self.committed = True
+
+
+@contextmanager
+def _capture_connection(conn):
+    yield conn
+
+
+class _FakeTotp:
+    def __init__(self, secret):
+        self.secret = secret
+
+    def provisioning_uri(self, name, issuer_name):
+        return f"otpauth://totp/{issuer_name}:{name}?secret={self.secret}"
+
+
+class _FakePyotp:
+    @staticmethod
+    def random_base32():
+        return "ABCDEFGHIJKLMNOP"
+
+    TOTP = _FakeTotp
+
+
+def test_append_strategy_log_uses_parent_exists_guard(monkeypatch):
+    conn = _CaptureConn()
+    monkeypatch.setattr(strategy_runtime_logs, "get_db_connection", lambda: _capture_connection(conn))
+
+    strategy_runtime_logs.append_strategy_log(3347, "info", "Strategy execution loop exited")
+
+    sql, params = conn.cursor_obj.calls[0]
+    assert "WHERE EXISTS" in sql
+    assert "qd_strategies_trading" in sql
+    assert params[0] == 3347
+    assert params[-1] == 3347
+    assert conn.committed
+    assert conn.cursor_obj.closed
+
+
+def test_mfa_start_setup_returns_user_id_not_missing_id(monkeypatch):
+    conn = _CaptureConn()
+    monkeypatch.setenv("MFA_ENABLED", "true")
+    monkeypatch.setattr(mfa_service.MfaService, "ensure_schema", lambda self: None)
+    monkeypatch.setattr(mfa_service, "get_db_connection", lambda: _capture_connection(conn))
+    monkeypatch.setattr(mfa_service, "encrypt_credential_blob", lambda secret: f"encrypted:{secret}")
+
+    service = mfa_service.MfaService()
+    monkeypatch.setattr(service, "_load_totp_libs", lambda: (_FakePyotp, object()))
+    monkeypatch.setattr(service, "_make_qr_data_url", lambda _qrcode, _uri: "data:image/png;base64,test")
+
+    result = service.start_setup(9383, "user@example.com")
+
+    sql, params = conn.cursor_obj.calls[0]
+    assert "RETURNING user_id" in sql
+    assert "RETURNING id" not in sql
+    assert params == (9383, "encrypted:ABCDEFGHIJKLMNOP")
+    assert result["secret"] == "ABCDEFGHIJKLMNOP"
+    assert conn.committed