agent功能开发增加MCP后端

mcp_database.py

@@ -0,0 +1,546 @@
+"""
+MySQL数据库查询模块
+提供股票财务数据查询功能
+"""
+
+import aiomysql
+import logging
+from typing import Dict, List, Any, Optional
+from datetime import datetime, date
+from decimal import Decimal
+import json
+
+logger = logging.getLogger(__name__)
+
+# MySQL连接配置
+MYSQL_CONFIG = {
+    'host': '222.128.1.157',
+    'port': 33060,
+    'user': 'root',
+    'password': 'Zzl5588161!',
+    'db': 'stock',
+    'charset': 'utf8mb4',
+    'autocommit': True
+}
+
+# 全局连接池
+_pool = None
+
+
+class DateTimeEncoder(json.JSONEncoder):
+    """JSON编码器，处理datetime和Decimal类型"""
+    def default(self, obj):
+        if isinstance(obj, (datetime, date)):
+            return obj.isoformat()
+        if isinstance(obj, Decimal):
+            return float(obj)
+        return super().default(obj)
+
+
+async def get_pool():
+    """获取MySQL连接池"""
+    global _pool
+    if _pool is None:
+        _pool = await aiomysql.create_pool(
+            host=MYSQL_CONFIG['host'],
+            port=MYSQL_CONFIG['port'],
+            user=MYSQL_CONFIG['user'],
+            password=MYSQL_CONFIG['password'],
+            db=MYSQL_CONFIG['db'],
+            charset=MYSQL_CONFIG['charset'],
+            autocommit=MYSQL_CONFIG['autocommit'],
+            minsize=1,
+            maxsize=10
+        )
+        logger.info("MySQL connection pool created")
+    return _pool
+
+
+async def close_pool():
+    """关闭MySQL连接池"""
+    global _pool
+    if _pool:
+        _pool.close()
+        await _pool.wait_closed()
+        _pool = None
+        logger.info("MySQL connection pool closed")
+
+
+def convert_row(row: Dict) -> Dict:
+    """转换数据库行，处理特殊类型"""
+    if not row:
+        return {}
+
+    result = {}
+    for key, value in row.items():
+        if isinstance(value, Decimal):
+            result[key] = float(value)
+        elif isinstance(value, (datetime, date)):
+            result[key] = value.isoformat()
+        else:
+            result[key] = value
+    return result
+
+
+async def get_stock_basic_info(seccode: str) -> Optional[Dict[str, Any]]:
+    """
+    获取股票基本信息
+
+    Args:
+        seccode: 股票代码
+
+    Returns:
+        股票基本信息字典
+    """
+    pool = await get_pool()
+
+    async with pool.acquire() as conn:
+        async with conn.cursor(aiomysql.DictCursor) as cursor:
+            query = """
+                SELECT
+                    SECCODE, SECNAME, ORGNAME,
+                    F001V as english_name,
+                    F003V as legal_representative,
+                    F004V as registered_address,
+                    F005V as office_address,
+                    F010D as establishment_date,
+                    F011V as website,
+                    F012V as email,
+                    F013V as phone,
+                    F015V as main_business,
+                    F016V as business_scope,
+                    F017V as company_profile,
+                    F030V as industry_level1,
+                    F032V as industry_level2,
+                    F034V as sw_industry_level1,
+                    F036V as sw_industry_level2,
+                    F026V as province,
+                    F028V as city,
+                    F041V as chairman,
+                    F042V as general_manager,
+                    UPDATE_DATE as update_date
+                FROM ea_baseinfo
+                WHERE SECCODE = %s
+                LIMIT 1
+            """
+
+            await cursor.execute(query, (seccode,))
+            result = await cursor.fetchone()
+
+            if result:
+                return convert_row(result)
+            return None
+
+
+async def get_stock_financial_index(
+    seccode: str,
+    start_date: Optional[str] = None,
+    end_date: Optional[str] = None,
+    limit: int = 10
+) -> List[Dict[str, Any]]:
+    """
+    获取股票财务指标
+
+    Args:
+        seccode: 股票代码
+        start_date: 开始日期 YYYY-MM-DD
+        end_date: 结束日期 YYYY-MM-DD
+        limit: 返回条数
+
+    Returns:
+        财务指标列表
+    """
+    pool = await get_pool()
+
+    async with pool.acquire() as conn:
+        async with conn.cursor(aiomysql.DictCursor) as cursor:
+            # 构建查询
+            query = """
+                SELECT
+                    SECCODE, SECNAME, ENDDATE, STARTDATE,
+                    F069D as report_year,
+                    F003N as eps,  -- 每股收益
+                    F004N as basic_eps,
+                    F008N as bps,  -- 每股净资产
+                    F014N as roe,  -- 净资产收益率
+                    F016N as roa,  -- 总资产报酬率
+                    F017N as net_profit_margin,  -- 净利润率
+                    F022N as receivable_turnover,  -- 应收账款周转率
+                    F023N as inventory_turnover,  -- 存货周转率
+                    F025N as total_asset_turnover,  -- 总资产周转率
+                    F041N as debt_ratio,  -- 资产负债率
+                    F042N as current_ratio,  -- 流动比率
+                    F043N as quick_ratio,  -- 速动比率
+                    F052N as revenue_growth,  -- 营业收入增长率
+                    F053N as profit_growth,  -- 净利润增长率
+                    F089N as revenue,  -- 营业收入
+                    F090N as operating_cost,  -- 营业成本
+                    F101N as net_profit,  -- 净利润
+                    F102N as net_profit_parent  -- 归母净利润
+                FROM ea_financialindex
+                WHERE SECCODE = %s
+            """
+
+            params = [seccode]
+
+            if start_date:
+                query += " AND ENDDATE >= %s"
+                params.append(start_date)
+
+            if end_date:
+                query += " AND ENDDATE <= %s"
+                params.append(end_date)
+
+            query += " ORDER BY ENDDATE DESC LIMIT %s"
+            params.append(limit)
+
+            await cursor.execute(query, params)
+            results = await cursor.fetchall()
+
+            return [convert_row(row) for row in results]
+
+
+async def get_stock_trade_data(
+    seccode: str,
+    start_date: Optional[str] = None,
+    end_date: Optional[str] = None,
+    limit: int = 30
+) -> List[Dict[str, Any]]:
+    """
+    获取股票交易数据
+
+    Args:
+        seccode: 股票代码
+        start_date: 开始日期 YYYY-MM-DD
+        end_date: 结束日期 YYYY-MM-DD
+        limit: 返回条数
+
+    Returns:
+        交易数据列表
+    """
+    pool = await get_pool()
+
+    async with pool.acquire() as conn:
+        async with conn.cursor(aiomysql.DictCursor) as cursor:
+            query = """
+                SELECT
+                    SECCODE, SECNAME, TRADEDATE,
+                    F002N as prev_close,  -- 昨日收盘价
+                    F003N as open_price,  -- 开盘价
+                    F005N as high_price,  -- 最高价
+                    F006N as low_price,  -- 最低价
+                    F007N as close_price,  -- 收盘价
+                    F004N as volume,  -- 成交量
+                    F011N as turnover,  -- 成交金额
+                    F009N as change_amount,  -- 涨跌额
+                    F010N as change_pct,  -- 涨跌幅
+                    F012N as turnover_rate,  -- 换手率
+                    F013N as amplitude,  -- 振幅
+                    F026N as pe_ratio,  -- 市盈率
+                    F020N as total_shares,  -- 总股本
+                    F021N as circulating_shares  -- 流通股本
+                FROM ea_trade
+                WHERE SECCODE = %s
+            """
+
+            params = [seccode]
+
+            if start_date:
+                query += " AND TRADEDATE >= %s"
+                params.append(start_date)
+
+            if end_date:
+                query += " AND TRADEDATE <= %s"
+                params.append(end_date)
+
+            query += " ORDER BY TRADEDATE DESC LIMIT %s"
+            params.append(limit)
+
+            await cursor.execute(query, params)
+            results = await cursor.fetchall()
+
+            return [convert_row(row) for row in results]
+
+
+async def get_stock_balance_sheet(
+    seccode: str,
+    start_date: Optional[str] = None,
+    end_date: Optional[str] = None,
+    limit: int = 8
+) -> List[Dict[str, Any]]:
+    """
+    获取资产负债表数据
+
+    Args:
+        seccode: 股票代码
+        start_date: 开始日期
+        end_date: 结束日期
+        limit: 返回条数
+
+    Returns:
+        资产负债表数据列表
+    """
+    pool = await get_pool()
+
+    async with pool.acquire() as conn:
+        async with conn.cursor(aiomysql.DictCursor) as cursor:
+            query = """
+                SELECT
+                    SECCODE, SECNAME, ENDDATE,
+                    F001D as report_year,
+                    F006N as cash,  -- 货币资金
+                    F009N as receivables,  -- 应收账款
+                    F015N as inventory,  -- 存货
+                    F019N as current_assets,  -- 流动资产合计
+                    F023N as long_term_investment,  -- 长期股权投资
+                    F025N as fixed_assets,  -- 固定资产
+                    F037N as noncurrent_assets,  -- 非流动资产合计
+                    F038N as total_assets,  -- 资产总计
+                    F039N as short_term_loan,  -- 短期借款
+                    F042N as payables,  -- 应付账款
+                    F052N as current_liabilities,  -- 流动负债合计
+                    F053N as long_term_loan,  -- 长期借款
+                    F060N as noncurrent_liabilities,  -- 非流动负债合计
+                    F061N as total_liabilities,  -- 负债合计
+                    F062N as share_capital,  -- 股本
+                    F063N as capital_reserve,  -- 资本公积
+                    F065N as retained_earnings,  -- 未分配利润
+                    F070N as total_equity  -- 所有者权益合计
+                FROM ea_asset
+                WHERE SECCODE = %s
+            """
+
+            params = [seccode]
+
+            if start_date:
+                query += " AND ENDDATE >= %s"
+                params.append(start_date)
+
+            if end_date:
+                query += " AND ENDDATE <= %s"
+                params.append(end_date)
+
+            query += " ORDER BY ENDDATE DESC LIMIT %s"
+            params.append(limit)
+
+            await cursor.execute(query, params)
+            results = await cursor.fetchall()
+
+            return [convert_row(row) for row in results]
+
+
+async def get_stock_cashflow(
+    seccode: str,
+    start_date: Optional[str] = None,
+    end_date: Optional[str] = None,
+    limit: int = 8
+) -> List[Dict[str, Any]]:
+    """
+    获取现金流量表数据
+
+    Args:
+        seccode: 股票代码
+        start_date: 开始日期
+        end_date: 结束日期
+        limit: 返回条数
+
+    Returns:
+        现金流量表数据列表
+    """
+    pool = await get_pool()
+
+    async with pool.acquire() as conn:
+        async with conn.cursor(aiomysql.DictCursor) as cursor:
+            query = """
+                SELECT
+                    SECCODE, SECNAME, ENDDATE, STARTDATE,
+                    F001D as report_year,
+                    F009N as operating_cash_inflow,  -- 经营活动现金流入
+                    F014N as operating_cash_outflow,  -- 经营活动现金流出
+                    F015N as net_operating_cashflow,  -- 经营活动现金流量净额
+                    F021N as investing_cash_inflow,  -- 投资活动现金流入
+                    F026N as investing_cash_outflow,  -- 投资活动现金流出
+                    F027N as net_investing_cashflow,  -- 投资活动现金流量净额
+                    F031N as financing_cash_inflow,  -- 筹资活动现金流入
+                    F035N as financing_cash_outflow,  -- 筹资活动现金流出
+                    F036N as net_financing_cashflow,  -- 筹资活动现金流量净额
+                    F039N as net_cash_increase,  -- 现金及现金等价物净增加额
+                    F044N as net_profit,  -- 净利润
+                    F046N as depreciation,  -- 固定资产折旧
+                    F060N as net_operating_cashflow_adjusted  -- 经营活动现金流量净额(补充)
+                FROM ea_cashflow
+                WHERE SECCODE = %s
+            """
+
+            params = [seccode]
+
+            if start_date:
+                query += " AND ENDDATE >= %s"
+                params.append(start_date)
+
+            if end_date:
+                query += " AND ENDDATE <= %s"
+                params.append(end_date)
+
+            query += " ORDER BY ENDDATE DESC LIMIT %s"
+            params.append(limit)
+
+            await cursor.execute(query, params)
+            results = await cursor.fetchall()
+
+            return [convert_row(row) for row in results]
+
+
+async def search_stocks_by_criteria(
+    industry: Optional[str] = None,
+    province: Optional[str] = None,
+    min_market_cap: Optional[float] = None,
+    max_market_cap: Optional[float] = None,
+    limit: int = 50
+) -> List[Dict[str, Any]]:
+    """
+    按条件搜索股票
+
+    Args:
+        industry: 行业名称
+        province: 省份
+        min_market_cap: 最小市值（亿元）
+        max_market_cap: 最大市值（亿元）
+        limit: 返回条数
+
+    Returns:
+        股票列表
+    """
+    pool = await get_pool()
+
+    async with pool.acquire() as conn:
+        async with conn.cursor(aiomysql.DictCursor) as cursor:
+            query = """
+                SELECT DISTINCT
+                    b.SECCODE,
+                    b.SECNAME,
+                    b.F030V as industry_level1,
+                    b.F032V as industry_level2,
+                    b.F034V as sw_industry_level1,
+                    b.F026V as province,
+                    b.F028V as city,
+                    b.F015V as main_business,
+                    t.F007N as latest_price,
+                    t.F010N as change_pct,
+                    t.F026N as pe_ratio,
+                    t.TRADEDATE as latest_trade_date
+                FROM ea_baseinfo b
+                LEFT JOIN (
+                    SELECT SECCODE, MAX(TRADEDATE) as max_date
+                    FROM ea_trade
+                    GROUP BY SECCODE
+                ) latest ON b.SECCODE = latest.SECCODE
+                LEFT JOIN ea_trade t ON b.SECCODE = t.SECCODE
+                    AND t.TRADEDATE = latest.max_date
+                WHERE 1=1
+            """
+
+            params = []
+
+            if industry:
+                query += " AND (b.F030V LIKE %s OR b.F032V LIKE %s OR b.F034V LIKE %s)"
+                pattern = f"%{industry}%"
+                params.extend([pattern, pattern, pattern])
+
+            if province:
+                query += " AND b.F026V = %s"
+                params.append(province)
+
+            if min_market_cap or max_market_cap:
+                # 市值 = 最新价 * 总股本 / 100000000（转换为亿元）
+                if min_market_cap:
+                    query += " AND (t.F007N * t.F020N / 100000000) >= %s"
+                    params.append(min_market_cap)
+
+                if max_market_cap:
+                    query += " AND (t.F007N * t.F020N / 100000000) <= %s"
+                    params.append(max_market_cap)
+
+            query += " ORDER BY t.TRADEDATE DESC LIMIT %s"
+            params.append(limit)
+
+            await cursor.execute(query, params)
+            results = await cursor.fetchall()
+
+            return [convert_row(row) for row in results]
+
+
+async def get_stock_comparison(
+    seccodes: List[str],
+    metric: str = "financial"
+) -> Dict[str, Any]:
+    """
+    股票对比分析
+
+    Args:
+        seccodes: 股票代码列表
+        metric: 对比指标类型 (financial/trade)
+
+    Returns:
+        对比数据
+    """
+    pool = await get_pool()
+
+    if not seccodes or len(seccodes) < 2:
+        return {"error": "至少需要2个股票代码进行对比"}
+
+    async with pool.acquire() as conn:
+        async with conn.cursor(aiomysql.DictCursor) as cursor:
+            placeholders = ','.join(['%s'] * len(seccodes))
+
+            if metric == "financial":
+                # 对比最新财务指标
+                query = f"""
+                    SELECT
+                        f.SECCODE, f.SECNAME, f.ENDDATE,
+                        f.F003N as eps,
+                        f.F008N as bps,
+                        f.F014N as roe,
+                        f.F017N as net_profit_margin,
+                        f.F041N as debt_ratio,
+                        f.F052N as revenue_growth,
+                        f.F053N as profit_growth,
+                        f.F089N as revenue,
+                        f.F101N as net_profit
+                    FROM ea_financialindex f
+                    INNER JOIN (
+                        SELECT SECCODE, MAX(ENDDATE) as max_date
+                        FROM ea_financialindex
+                        WHERE SECCODE IN ({placeholders})
+                        GROUP BY SECCODE
+                    ) latest ON f.SECCODE = latest.SECCODE
+                        AND f.ENDDATE = latest.max_date
+                """
+            else:  # trade
+                # 对比最新交易数据
+                query = f"""
+                    SELECT
+                        t.SECCODE, t.SECNAME, t.TRADEDATE,
+                        t.F007N as close_price,
+                        t.F010N as change_pct,
+                        t.F012N as turnover_rate,
+                        t.F026N as pe_ratio,
+                        t.F020N as total_shares,
+                        t.F021N as circulating_shares
+                    FROM ea_trade t
+                    INNER JOIN (
+                        SELECT SECCODE, MAX(TRADEDATE) as max_date
+                        FROM ea_trade
+                        WHERE SECCODE IN ({placeholders})
+                        GROUP BY SECCODE
+                    ) latest ON t.SECCODE = latest.SECCODE
+                        AND t.TRADEDATE = latest.max_date
+                """
+
+            await cursor.execute(query, seccodes)
+            results = await cursor.fetchall()
+
+            return {
+                "comparison_type": metric,
+                "stocks": [convert_row(row) for row in results]
+            }