sanguo_quant_live/value-investing/super_financial_agent.py

#!/usr/bin/env python3
"""
超级财务智能体 - 并行财务因子计算引擎
启动时间：2026-03-21 17:45
截止时间：18:00
"""

import sys
import os
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor, as_completed
import warnings
warnings.filterwarnings('ignore')

class SuperFinancialAgent:
    """超级财务智能体"""
    
    def __init__(self):
        self.start_time = datetime.now()
        self.cpu_cores = min(10, mp.cpu_count())
        print(f"⚡ 超级财务智能体启动！")
        print(f"🕐 启动时间: {self.start_time.strftime('%H:%M:%S')}")
        print(f"🎯 目标时间: 18:00")
        print(f"⏰ 剩余时间: {15}分钟")
        print(f"💻 CPU核心: {self.cpu_cores}核并行")
        
    def calculate_valuation_factors(self, stock_data):
        """计算估值因子（核心1-3）"""
        factors = {}
        
        # 1. 市盈率因子
        if 'pe_ratio' in stock_data.columns:
            factors['pe_rank'] = stock_data['pe_ratio'].rank(pct=True)
            factors['pe_zscore'] = (stock_data['pe_ratio'] - stock_data['pe_ratio'].mean()) / stock_data['pe_ratio'].std()
        
        # 2. 市净率因子
        if 'pb_ratio' in stock_data.columns:
            factors['pb_rank'] = stock_data['pb_ratio'].rank(pct=True)
            factors['pb_zscore'] = (stock_data['pb_ratio'] - stock_data['pb_ratio'].mean()) / stock_data['pb_ratio'].std()
        
        # 3. 市销率因子
        if 'ps_ratio' in stock_data.columns:
            factors['ps_rank'] = stock_data['ps_ratio'].rank(pct=True)
            factors['ps_zscore'] = (stock_data['ps_ratio'] - stock_data['ps_ratio'].mean()) / stock_data['ps_ratio'].std()
        
        return factors
    
    def calculate_quality_factors(self, stock_data):
        """计算质量因子（核心4-6）"""
        factors = {}
        
        # 4. ROE因子
        if 'roe' in stock_data.columns:
            factors['roe_rank'] = stock_data['roe'].rank(pct=True)
            factors['roe_stability'] = stock_data['roe'].rolling(5).std()
        
        # 5. 毛利率因子
        if 'gross_margin' in stock_data.columns:
            factors['gross_margin_rank'] = stock_data['gross_margin'].rank(pct=True)
            factors['margin_stability'] = stock_data['gross_margin'].rolling(5).std()
        
        # 6. 现金流因子
        if 'free_cash_flow' in stock_data.columns:
            factors['fcf_rank'] = stock_data['free_cash_flow'].rank(pct=True)
            factors['fcf_yield'] = stock_data['free_cash_flow'] / stock_data['market_cap']
        
        return factors
    
    def calculate_growth_factors(self, stock_data):
        """计算成长因子（核心7-8）"""
        factors = {}
        
        # 7. 营收增长因子
        if 'revenue' in stock_data.columns:
            revenue_growth = stock_data['revenue'].pct_change(periods=4)
            factors['revenue_growth_rank'] = revenue_growth.rank(pct=True)
        
        # 8. 盈利增长因子
        if 'net_profit' in stock_data.columns:
            profit_growth = stock_data['net_profit'].pct_change(periods=4)
            factors['profit_growth_rank'] = profit_growth.rank(pct=True)
        
        return factors
    
    def calculate_risk_factors(self, stock_data):
        """计算风险因子（核心9-10）"""
        factors = {}
        
        # 9. 波动率因子
        if 'close' in stock_data.columns:
            volatility = stock_data['close'].rolling(20).std()
            factors['volatility_rank'] = volatility.rank(pct=True)
        
        # 10. 流动性因子
        if 'volume' in stock_data.columns:
            avg_volume = stock_data['volume'].rolling(20).mean()
            factors['liquidity_rank'] = avg_volume.rank(pct=True)
        
        return factors
    
    def parallel_factor_calculation(self, stock_data):
        """并行计算所有因子"""
        print(f"🔢 开始并行因子计算...")
        
        # 准备任务
        tasks = [
            (self.calculate_valuation_factors, stock_data),
            (self.calculate_quality_factors, stock_data),
            (self.calculate_growth_factors, stock_data),
            (self.calculate_risk_factors, stock_data)
        ]
        
        # 并行计算
        all_factors = {}
        with ProcessPoolExecutor(max_workers=self.cpu_cores) as executor:
            future_to_task = {executor.submit(func, data): (func.__name__, data) for func, data in tasks}
            
            for future in as_completed(future_to_task):
                task_name, _ = future_to_task[future]
                try:
                    factors = future.result()
                    all_factors.update(factors)
                    print(f"✅ {task_name} 计算完成")
                except Exception as e:
                    print(f"❌ {task_name} 计算失败: {e}")
        
        return all_factors

def main():
    """主函数"""
    agent = SuperFinancialAgent()
    
    # 1. 生成模拟数据（实际项目中从数据源获取）
    print(f"\n📊 生成模拟财务数据...")
    np.random.seed(42)
    n_stocks = 3000
    
    stock_data = pd.DataFrame({
        'stock_code': [f'{i:06d}.XSHE' for i in range(1, n_stocks + 1)],
        'pe_ratio': np.random.uniform(5, 50, n_stocks),
        'pb_ratio': np.random.uniform(0.5, 5, n_stocks),
        'ps_ratio': np.random.uniform(0.5, 10, n_stocks),
        'roe': np.random.uniform(0.05, 0.3, n_stocks),
        'gross_margin': np.random.uniform(0.2, 0.6, n_stocks),
        'free_cash_flow': np.random.uniform(1e6, 1e9, n_stocks),
        'market_cap': np.random.uniform(1e8, 1e11, n_stocks),
        'revenue': np.random.uniform(1e7, 1e10, n_stocks),
        'net_profit': np.random.uniform(1e6, 1e9, n_stocks),
        'close': np.random.uniform(10, 100, n_stocks),
        'volume': np.random.uniform(1e5, 1e7, n_stocks)
    })
    
    print(f"✅ 生成 {n_stocks} 只股票财务数据")
    
    # 2. 并行计算因子
    factors = agent.parallel_factor_calculation(stock_data)
    
    # 3. 合并因子数据
    for factor_name, factor_values in factors.items():
        stock_data[factor_name] = factor_values
    
    # 4. 计算综合价值得分
    print(f"\n🎯 计算综合价值得分...")
    
    # 价值因子（越低越好）
    value_factors = ['pe_rank', 'pb_rank', 'ps_rank']
    value_score = stock_data[value_factors].mean(axis=1)
    
    # 质量因子（越高越好）
    quality_factors = ['roe_rank', 'gross_margin_rank', 'fcf_rank']
    quality_score = stock_data[quality_factors].mean(axis=1)
    
    # 综合得分：价值得分（高） + 质量得分（高）
    stock_data['value_quality_score'] = (1 - value_score) * 0.6 + quality_score * 0.4
    stock_data['value_quality_rank'] = stock_data['value_quality_score'].rank(ascending=False, pct=True)
    
    # 5. 输出结果
    print(f"\n📈 计算完成！")
    print(f"⏰ 耗时: {(datetime.now() - agent.start_time).total_seconds():.2f}秒")
    print(f"📊 总因子数: {len(factors)}个")
    print(f"🏆 综合价值得分计算完成")
    
    # 显示前10名
    top_stocks = stock_data.nlargest(10, 'value_quality_score')[['stock_code', 'value_quality_score', 'value_quality_rank']]
    print(f"\n🏅 价值投资前10名:")
    print(top_stocks.to_string(index=False))
    
    return stock_data

if __name__ == "__main__":
    stock_data = main()