4、数据库数据源连接：MySQL、PostgreSQL连接配置、SQL查询基础、连接池管理

数据库连接，是金融Agent的「血管」。数据流不通，再聪明的Agent也是白搭。

我这些年做金融数据工程，打交道最多的就是MySQL和PostgreSQL。一个像老黄牛，稳定可靠；一个像瑞士军刀，功能全面。今天咱们就把这两个家伙的连接配置、查询技巧和连接池管理，一次性讲透。

4.1 连接配置：别让Agent「迷路」

连接数据库，说白了就是告诉Agent三件事：去哪连、怎么连、连上后用什么身份。

4.1.1 MySQL 连接配置

我个人习惯用 mysql-connector-python，轻量且稳定。配置参数看起来多，其实核心就几个：

import mysql.connector

config = {
    'host': '192.168.1.100',      # 数据库服务器IP
    'port': 3306,                  # MySQL默认端口
    'user': 'fin_agent',           # 用户名
    'password': 'Str0ng!Pass',     # 密码
    'database': 'market_data',     # 数据库名
    'charset': 'utf8mb4',          # 字符集，支持emoji
    'connect_timeout': 10,         # 连接超时（秒）
    'autocommit': True             # 自动提交事务
}

conn = mysql.connector.connect(**config)

4.1.2 PostgreSQL 连接配置

PostgreSQL 我推荐用 psycopg2，它是PG生态里的「老大哥」。配置方式类似，但有几个PG特有的参数：

import psycopg2

conn = psycopg2.connect(
    host='192.168.1.101',
    port=5432,
    user='fin_agent',
    password='Str0ng!Pass',
    dbname='market_data',
    connect_timeout=10,
    sslmode='require',          # 金融数据建议开启SSL
    keepalives=1,               # 保持心跳
    keepalives_idle=30,         # 空闲30秒后发心跳
    keepalives_interval=10,     # 心跳间隔10秒
    keepalives_count=5          # 最多重试5次
)

4.2 SQL查询基础：Agent的「语言」

Agent要跟数据库对话，SQL就是它的语言。别觉得SQL简单，金融场景下的查询，讲究的是精准、高效、安全。

4.2.1 基础查询模板

先来个最常用的——查某只股票的历史行情：

import pandas as pd

query = """
SELECT 
    trade_date,
    open_price,
    close_price,
    high_price,
    low_price,
    volume
FROM stock_daily
WHERE stock_code = '600519'
  AND trade_date BETWEEN '2024-01-01' AND '2024-12-31'
ORDER BY trade_date ASC
"""

df = pd.read_sql(query, conn)

嗯，这里要注意：永远不要用字符串拼接SQL。你想想看，如果用户输入的是 600519'; DROP TABLE stock_daily; --，那你的数据库就没了。用参数化查询才是正道：

query = """
SELECT * FROM stock_daily
WHERE stock_code = %s
  AND trade_date BETWEEN %s AND %s
"""
params = ('600519', '2024-01-01', '2024-12-31')
df = pd.read_sql(query, conn, params=params)

4.2.2 金融场景常用查询模式

做金融Agent，下面几种查询模式你一定会遇到：

场景	SQL 示例	说明
查最新行情	SELECT * FROM quotes ORDER BY ts DESC LIMIT 1	获取最新一条数据
查日频数据	SELECT date, close FROM daily WHERE code='000001'	按日期筛选
聚合统计	SELECT AVG(close), MAX(high) FROM daily	计算均值、极值
多表关联	SELECT a.*, b.name FROM trades a JOIN stocks b ON a.code=b.code	关联股票基本信息
窗口函数	SELECT close, LAG(close,1) OVER(ORDER BY date) AS prev_close	计算前一日收盘价

4.3 连接池管理：别让Agent「累死」

金融Agent通常要同时处理几十个任务。如果每个任务都新建一个数据库连接，那数据库服务器很快就会被「撑爆」。连接池就是解决这个问题的——复用连接，按需分配。

4.3.1 为什么需要连接池？

我举个例子你就明白了。假设你的Agent要同时查询100只股票的实时行情：

• 不用连接池：每次查询都新建连接，100次查询就要建立100次TCP握手、100次身份验证。耗时至少增加5倍。
• 用连接池：预先建立10个连接，100次查询排队使用这10个连接。连接复用，性能飙升。

说白了，连接池就是「预分配、重复用」——跟银行柜台一样，不用每次办业务都重新招一个柜员。

4.3.2 连接池配置实战

Python里最常用的连接池是 SQLAlchemy 自带的 QueuePool。配置起来很简单：

from sqlalchemy import create_engine

# MySQL 连接池
engine_mysql = create_engine(
    'mysql+mysqlconnector://fin_agent:Str0ng!Pass@192.168.1.100:3306/market_data',
    pool_size=10,           # 池中保持10个连接
    max_overflow=20,        # 最多额外创建20个连接（应对突发流量）
    pool_timeout=30,        # 等待连接的超时时间（秒）
    pool_recycle=3600,      # 连接每1小时回收重建（防止连接失效）
    pool_pre_ping=True      # 每次使用前检查连接是否存活
)

# PostgreSQL 连接池
engine_pg = create_engine(
    'postgresql+psycopg2://fin_agent:Str0ng!Pass@192.168.1.101:5432/market_data',
    pool_size=10,
    max_overflow=20,
    pool_timeout=30,
    pool_recycle=3600,
    pool_pre_ping=True
)

4.3.3 连接池参数调优

参数怎么调？没有银弹，但有经验公式：

参数	推荐值	我的经验
pool_size	CPU核心数 × 2	8核机器设16，16核机器设32。别贪多，连接太多反而增加数据库负担。
max_overflow	pool_size × 2	应对突发查询高峰。比如开盘前5分钟，查询量会暴增3倍。
pool_recycle	3600秒（1小时）	数据库默认8小时断开空闲连接，我们提前回收，避免「断连」报错。
pool_timeout	30秒	超过30秒拿不到连接，说明系统过载了，直接报错比死等强。

4.4 知识体系总览

这一章的内容比较多，我画了张图帮你梳理一下核心逻辑：

这张图把三大模块串起来了。你想想看，连接配置是「地基」，SQL查询是「工具」，连接池是「调度中心」——三者缺一不可。