v3.0.2

Signed-off-by: Dinger <quantdinger@gmail.com>

Author: quantdinger-lab

backend_api_python/app/routes/market.py

@@ -3,6 +3,7 @@
 Provides watchlist, market metadata, symbol search, and pricing helpers for the frontend.
 """
 from flask import Blueprint, request, jsonify, g
+import os
 import traceback
 import json
 import time
@@ -27,8 +28,17 @@
 kline_service = KlineService()
 cache = CacheManager()
 
-# Thread pool for parallel price fetching
-executor = ThreadPoolExecutor(max_workers=10)
+# Thread pool for parallel price fetching.
+# Each task may grab a DB connection, so keep this well below DB_POOL_MAX.
+# Tunable via MARKET_EXECUTOR_WORKERS env.
+def _market_executor_workers() -> int:
+    try:
+        v = int(os.getenv("MARKET_EXECUTOR_WORKERS", "6"))
+        return v if v > 0 else 6
+    except Exception:
+        return 6
+
+executor = ThreadPoolExecutor(max_workers=_market_executor_workers())
 
 def _now_ts() -> int:
     return int(time.time())

backend_api_python/app/routes/portfolio.py

@@ -4,6 +4,7 @@
 """
 from flask import Blueprint, request, jsonify, g
 from datetime import date, datetime, timezone
+import os
 import json
 import traceback
 import time
@@ -24,9 +25,18 @@
 kline_service = KlineService()
 cache = CacheManager()
 
-# Thread pool for parallel price fetching
-# Lower concurrency to avoid triggering API limits (especially for forex/US stocks)
-executor = ThreadPoolExecutor(max_workers=3)
+# Thread pool for parallel price fetching.
+# Lower concurrency to avoid triggering API limits (especially for forex/US
+# stocks) and to keep pressure off the DB connection pool.
+# Tunable via PORTFOLIO_EXECUTOR_WORKERS env.
+def _portfolio_executor_workers() -> int:
+    try:
+        v = int(os.getenv("PORTFOLIO_EXECUTOR_WORKERS", "3"))
+        return v if v > 0 else 3
+    except Exception:
+        return 3
+
+executor = ThreadPoolExecutor(max_workers=_portfolio_executor_workers())
 
 # Request interval (seconds) to avoid too frequent requests
 REQUEST_INTERVAL = 0.3

backend_api_python/app/utils/db_postgres.py

@@ -3,8 +3,15 @@
 
 Supports multi-user mode with connection pooling.
 Provides placeholder conversion for backward compatibility with legacy code.
+
+Pool tuning (all via env, safe defaults):
+    DB_POOL_MIN               minconn                       default 5
+    DB_POOL_MAX               maxconn                       default 50
+    DB_POOL_ACQUIRE_TIMEOUT   seconds to wait on exhaustion default 10
+    DB_POOL_HEALTH_CHECK      "true" / "false"              default "true"
 """
 import os
+import time
 import threading
 from typing import Optional, Any, List, Dict
 from contextlib import contextmanager
@@ -16,6 +23,7 @@
 try:
     import psycopg2
     from psycopg2 import pool
+    from psycopg2 import OperationalError, InterfaceError
     from psycopg2.extras import RealDictCursor
     HAS_PSYCOPG2 = True
 except ImportError:
@@ -27,6 +35,27 @@
 _pool_lock = threading.Lock()
 
 
+def _env_int(key: str, default: int) -> int:
+    try:
+        v = int(os.getenv(key, str(default)))
+        return v if v > 0 else default
+    except Exception:
+        return default
+
+
+def _env_bool(key: str, default: bool) -> bool:
+    v = os.getenv(key)
+    if v is None:
+        return default
+    return v.strip().lower() in ("1", "true", "yes", "on")
+
+
+DB_POOL_MIN = _env_int("DB_POOL_MIN", 5)
+DB_POOL_MAX = _env_int("DB_POOL_MAX", 50)
+DB_POOL_ACQUIRE_TIMEOUT = _env_int("DB_POOL_ACQUIRE_TIMEOUT", 10)
+DB_POOL_HEALTH_CHECK = _env_bool("DB_POOL_HEALTH_CHECK", True)
+
+
 def _get_database_url() -> str:
     """Get database connection URL from environment"""
     return os.getenv('DATABASE_URL', '').strip()
@@ -99,23 +128,112 @@ def _get_connection_pool():
 
         try:
             _connection_pool = pool.ThreadedConnectionPool(
-                minconn=2,
-                maxconn=20,
+                minconn=DB_POOL_MIN,
+                maxconn=DB_POOL_MAX,
                 host=params.get('host', 'localhost'),
                 port=params.get('port', 5432),
                 user=params.get('user', 'quantdinger'),
                 password=params.get('password', ''),
                 dbname=params.get('dbname', 'quantdinger'),
                 connect_timeout=10,
+                # Apply timezone at connection establishment so we don't need
+                # per-checkout SET TIME ZONE (which left connections in an
+                # "idle in transaction" state when no explicit commit/rollback
+                # followed).  keepalives keep dead sockets from lingering in
+                # the pool when the PG side or a NAT drops them.
+                options="-c timezone=UTC",
+                keepalives=1,
+                keepalives_idle=30,
+                keepalives_interval=10,
+                keepalives_count=3,
+            )
+            logger.info(
+                f"PostgreSQL connection pool created: "
+                f"{params.get('host')}:{params.get('port')}/{params.get('dbname')} "
+                f"(min={DB_POOL_MIN}, max={DB_POOL_MAX}, "
+                f"acquire_timeout={DB_POOL_ACQUIRE_TIMEOUT}s, "
+                f"health_check={DB_POOL_HEALTH_CHECK})"
             )
-            logger.info(f"PostgreSQL connection pool created: {params.get('host')}:{params.get('port')}/{params.get('dbname')}")
         except Exception as e:
             logger.error(f"Failed to create PostgreSQL connection pool: {e}")
             raise
 
         return _connection_pool
 
 
+def _is_connection_healthy(conn) -> bool:
+    """Quick health check: make sure the connection is not closed and can
+    actually round-trip a trivial query.  Used only when DB_POOL_HEALTH_CHECK
+    is on, since SELECT 1 adds a small latency.
+    """
+    if conn is None:
+        return False
+    # psycopg2 sets .closed to nonzero when the connection is closed.
+    if getattr(conn, "closed", 0):
+        return False
+    try:
+        cur = conn.cursor()
+        cur.execute("SELECT 1")
+        cur.fetchone()
+        cur.close()
+        return True
+    except Exception:
+        return False
+
+
+def _acquire_conn_with_wait(pg_pool):
+    """Wrapper around pg_pool.getconn() that waits up to
+    DB_POOL_ACQUIRE_TIMEOUT seconds instead of failing immediately when the
+    pool is exhausted (psycopg2's default behaviour).  Also performs a
+    lightweight health check on the returned connection and discards dead
+    connections back to the pool to let PG reopen fresh ones.
+    """
+    if not HAS_PSYCOPG2:
+        raise RuntimeError("psycopg2 is not installed. Cannot use PostgreSQL.")
+
+    deadline = time.monotonic() + max(1, DB_POOL_ACQUIRE_TIMEOUT)
+    backoff = 0.05  # start at 50ms
+    last_err: Optional[Exception] = None
+    warned = False
+    while True:
+        try:
+            conn = pg_pool.getconn()
+        except pool.PoolError as e:
+            last_err = e
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                logger.error(
+                    f"PostgreSQL pool exhausted: all {DB_POOL_MAX} connections are in use "
+                    f"and waiting {DB_POOL_ACQUIRE_TIMEOUT}s did not free any. "
+                    f"Consider raising DB_POOL_MAX or investigating long-running queries."
+                )
+                raise
+            if not warned:
+                logger.warning(
+                    f"PostgreSQL pool exhausted ({DB_POOL_MAX} in use); "
+                    f"waiting up to {DB_POOL_ACQUIRE_TIMEOUT}s for a slot..."
+                )
+                warned = True
+            time.sleep(min(backoff, max(0.0, remaining)))
+            backoff = min(backoff * 2, 0.5)
+            continue
+
+        if DB_POOL_HEALTH_CHECK and not _is_connection_healthy(conn):
+            # Drop the dead connection and let the pool create a new one on
+            # next attempt.  putconn(close=True) asks the pool to discard it.
+            try:
+                pg_pool.putconn(conn, close=True)
+            except Exception:
+                pass
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise last_err or RuntimeError("DB pool returned only dead connections")
+            time.sleep(min(backoff, max(0.0, remaining)))
+            continue
+
+        return conn
+
+
 class PostgresCursor:
     """PostgreSQL cursor wrapper with placeholder conversion for backward compatibility"""
 
@@ -214,62 +332,65 @@ def rollback(self):
         self._conn.rollback()
 
     def close(self):
-        """Return connection to pool"""
+        """Return connection to pool.  Broken connections are discarded so
+        we don't poison the pool with closed sockets.
+        """
         if self._pool and self._conn:
             try:
-                self._pool.putconn(self._conn)
+                broken = bool(getattr(self._conn, "closed", 0))
+                self._pool.putconn(self._conn, close=broken)
             except Exception as e:
                 logger.warning(f"Failed to return connection to pool: {e}")
 
 
-def _ensure_session_utc(conn) -> None:
-    """每条连接检出后固定为 UTC，与 API 序列化约定一致。"""
-    try:
-        cur = conn.cursor()
-        cur.execute("SET TIME ZONE 'UTC'")
-        cur.close()
-    except Exception as e:
-        logger.warning(f"Could not SET TIME ZONE UTC on connection: {e}")
-
-
 @contextmanager
 def get_pg_connection():
     """
-    Get PostgreSQL database connection (Context Manager)
+    Get PostgreSQL database connection (Context Manager).
+
+    Uses _acquire_conn_with_wait so a momentary pool exhaustion does not
+    immediately fail the request; we wait up to DB_POOL_ACQUIRE_TIMEOUT
+    seconds for a connection to be released.
     """
-    pool = _get_connection_pool()
+    pg_pool = _get_connection_pool()
     conn = None
+    broken = False
     try:
-        conn = pool.getconn()
-        _ensure_session_utc(conn)
+        conn = _acquire_conn_with_wait(pg_pool)
         pg_conn = PostgresConnection(conn)
         yield pg_conn
     except Exception as e:
-        if conn:
+        if conn is not None:
             try:
                 conn.rollback()
             except Exception:
                 pass
-        # 记录更详细的错误信息
+            # If the connection itself died mid-request, discard it instead
+            # of returning it to the pool.
+            if isinstance(e, (OperationalError, InterfaceError)) or getattr(conn, "closed", 0):
+                broken = True
         error_msg = str(e) if e else repr(e)
         error_type = type(e).__name__
         logger.error(f"PostgreSQL operation error ({error_type}): {error_msg}", exc_info=True)
         raise
     finally:
-        if conn:
+        if conn is not None:
             try:
-                pool.putconn(conn)
+                pg_pool.putconn(conn, close=broken)
             except Exception:
                 pass
 
 
 def get_pg_connection_sync() -> PostgresConnection:
     """
-    Get connection synchronously (caller must close)
+    Get connection synchronously (caller must close).
+
+    NOTE: this function leaks its connection if the caller forgets to call
+    `.close()`.  Prefer `get_pg_connection()` (context manager) whenever
+    possible.
     """
-    pool = _get_connection_pool()
-    conn = pool.getconn()
-    _ensure_session_utc(conn)
+    pg_pool = _get_connection_pool()
+    conn = _acquire_conn_with_wait(pg_pool)
     return PostgresConnection(conn)
 
 

backend_api_python/env.example

@@ -22,6 +22,28 @@ DATABASE_URL=postgresql://quantdinger:quantdinger123@postgres:5432/quantdinger
 FRONTEND_URL=http://localhost:8888
 ENABLE_REGISTRATION=true
 
+# =========================
+# Database connection pool (psycopg2 ThreadedConnectionPool)
+# =========================
+# Tune these if you see `psycopg2.pool.PoolError: connection pool exhausted`
+# or if you run many trading bots / portfolios concurrently.
+# Make sure PG `max_connections` (docker-compose: PG_MAX_CONNECTIONS) is
+# comfortably larger than DB_POOL_MAX.
+DB_POOL_MIN=5
+DB_POOL_MAX=50
+DB_POOL_ACQUIRE_TIMEOUT=10
+DB_POOL_HEALTH_CHECK=true
+
+# Route-level parallel fetch executors.  Each worker may hold one DB
+# connection, so keep MARKET_EXECUTOR_WORKERS + PORTFOLIO_EXECUTOR_WORKERS
+# well below DB_POOL_MAX.
+MARKET_EXECUTOR_WORKERS=6
+PORTFOLIO_EXECUTOR_WORKERS=3
+
+# Gunicorn worker/thread model
+GUNICORN_WORKERS=1
+GUNICORN_THREADS=8
+
 # =========================
 # AI / LLM (choose one provider)
 # =========================

docker-compose.yml

@@ -33,6 +33,15 @@ services:
       POSTGRES_USER: ${POSTGRES_USER:-quantdinger}
       POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-quantdinger123}
       TZ: ${TZ:-Asia/Shanghai}
+    # Raise PostgreSQL's max_connections so the backend pool (default
+    # DB_POOL_MAX=50) plus psql/admin connections never exceed PG's own
+    # limit.  PG default is 100; 150 gives headroom for future scaling.
+    command:
+      - "postgres"
+      - "-c"
+      - "max_connections=${PG_MAX_CONNECTIONS:-150}"
+      - "-c"
+      - "shared_buffers=${PG_SHARED_BUFFERS:-256MB}"
     volumes:
       - postgres_data:/var/lib/postgresql/data
       - ./backend_api_python/migrations/init.sql:/docker-entrypoint-initdb.d/01-init.sql
@@ -96,6 +105,19 @@ services:
       - REDIS_HOST=redis
       - REDIS_PORT=6379
       - CACHE_ENABLED=true
+      # Database connection pool tuning (see app/utils/db_postgres.py).
+      # Defaults are safe for ~50 concurrent DB users; raise if you run many
+      # bots or see `connection pool exhausted` errors.
+      - DB_POOL_MIN=${DB_POOL_MIN:-5}
+      - DB_POOL_MAX=${DB_POOL_MAX:-50}
+      - DB_POOL_ACQUIRE_TIMEOUT=${DB_POOL_ACQUIRE_TIMEOUT:-10}
+      - DB_POOL_HEALTH_CHECK=${DB_POOL_HEALTH_CHECK:-true}
+      # Route-level executor tuning (each worker may use one DB connection).
+      - MARKET_EXECUTOR_WORKERS=${MARKET_EXECUTOR_WORKERS:-6}
+      - PORTFOLIO_EXECUTOR_WORKERS=${PORTFOLIO_EXECUTOR_WORKERS:-3}
+      # Gunicorn concurrency (threads within the single worker process).
+      - GUNICORN_WORKERS=${GUNICORN_WORKERS:-1}
+      - GUNICORN_THREADS=${GUNICORN_THREADS:-8}
     networks:
       - quantdinger-network
     healthcheck:

docs/CHANGELOG.md

@@ -4,7 +4,7 @@ This document records version updates, new features, bug fixes, and database mig
 
 ---
 
-## V3.0.4 (2026-04-11) — 多语言文件全量补齐(AI 自动翻译)
+## V3.0.2 (2026-04-11) — 多语言文件全量补齐(AI 自动翻译)
 
 ### 🌍 i18n
 
@@ -52,7 +52,7 @@ This document records version updates, new features, bug fixes, and database mig
 
 ---
 
-## V3.0.3 (2026-04-11) — 交易机器人全链路修复(Grid / Martingale / Trend / DCA)
+## V3.0.2 (2026-04-11) — 交易机器人全链路修复(Grid / Martingale / Trend / DCA)
 
 ### 🐛 Bug Fixes — 交易机器人