Rename app.py to analyze_corporate_finance.py

analyze_corporate_finance.py

@@ -0,0 +1,475 @@
+from smolagents import CodeAgent, DuckDuckGoSearchTool, HfApiModel, load_tool, tool
+import datetime
+import requests
+import pytz
+import yaml
+import pandas as pd
+import matplotlib.pyplot as plt
+import seaborn as sns
+import numpy as np
+from datetime import datetime, timedelta
+from typing import Dict, List, Optional, Tuple
+import io
+import base64
+from enum import Enum
+from tools.final_answer import FinalAnswerTool
+from Gradio_UI import GradioUI
+
+# 基础工具定义
+@tool
+def my_custom_tool(arg1: str, arg2: int) -> str:
+    """A tool that does nothing yet 
+    Args:
+        arg1: the first argument
+        arg2: the second argument
+    """
+    return "What magic will you build ?"
+
+@tool
+def get_current_time_in_timezone(timezone: str) -> str:
+    """A tool that fetches the current local time in a specified timezone.
+    Args:
+        timezone: A string representing a valid timezone (e.g., 'America/New_York').
+    """
+    try:
+        tz = pytz.timezone(timezone)
+        local_time = datetime.datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S")
+        return f"The current local time in {timezone} is: {local_time}"
+    except Exception as e:
+        return f"Error fetching time for timezone '{timezone}': {str(e)}"
+
+# 公司收支数据分析工具
+class AnalysisType(Enum):
+    OVERVIEW = "overview"
+    TREND = "trend"
+    CATEGORY = "category"
+    BUDGET = "budget"
+    CASH_FLOW = "cash_flow"
+    DEPARTMENT = "department"
+
+@tool
+def analyze_corporate_finance(data_source: str, 
+                             analysis_type: str = "overview",
+                             period: Optional[str] = None,
+                             department: Optional[str] = None,
+                             export_format: str = "text") -> str:
+    """
+    公司收支数据分析工具 - 专业财务分析系统
+    
+    Args:
+        data_source: 数据源路径(CSV/Excel)或JSON格式的财务数据
+        analysis_type: 分析类型 - overview, trend, category, budget, cash_flow, department
+        period: 时间周期筛选，如 '2024-Q1:2024-Q4' 或 'last_12_months'
+        department: 部门筛选，如 '销售部', '技术部'
+        export_format: 输出格式 - text, detailed, visual
+    """
+    try:
+        # 加载和预处理数据
+        df = load_financial_data(data_source)
+        df = preprocess_financial_data(df)
+        
+        # 应用筛选条件
+        df = apply_filters(df, period, department)
+        
+        if len(df) == 0:
+            return "没有找到符合条件的数据，请检查筛选条件"
+        
+        # 执行分析
+        if analysis_type == AnalysisType.OVERVIEW.value:
+            result = generate_financial_overview(df)
+        elif analysis_type == AnalysisType.TREND.value:
+            result = analyze_financial_trends(df)
+        elif analysis_type == AnalysisType.CATEGORY.value:
+            result = analyze_category_performance(df)
+        elif analysis_type == AnalysisType.BUDGET.value:
+            result = analyze_budget_variance(df)
+        elif analysis_type == AnalysisType.CASH_FLOW.value:
+            result = analyze_cash_flow(df)
+        elif analysis_type == AnalysisType.DEPARTMENT.value:
+            result = analyze_department_performance(df)
+        else:
+            return "不支持的分析类型，请选择: overview, trend, category, budget, cash_flow, department"
+        
+        # 根据输出格式调整结果
+        if export_format == "visual":
+            chart_url = generate_financial_charts(df, analysis_type)
+            return f"{result}\n\n![财务分析图表]({chart_url})"
+        elif export_format == "detailed":
+            return generate_detailed_report(df, analysis_type)
+        else:
+            return result
+            
+    except Exception as e:
+        return f"财务分析过程中出现错误: {str(e)}"
+
+def load_financial_data(data_source: str) -> pd.DataFrame:
+    """加载财务数据，支持多种格式"""
+    try:
+        if data_source.endswith('.csv'):
+            return pd.read_csv(data_source)
+        elif data_source.endswith(('.xlsx', '.xls')):
+            return pd.read_excel(data_source)
+        else:
+            # 尝试解析为JSON
+            return pd.read_json(io.StringIO(data_source))
+    except Exception as e:
+        raise Exception(f"数据加载失败: {str(e)}")
+
+def preprocess_financial_data(df: pd.DataFrame) -> pd.DataFrame:
+    """财务数据清洗和预处理"""
+    # 基本列名标准化
+    column_mapping = {
+        '日期': 'date', '时间': 'date', '交易日期': 'date',
+        '金额': 'amount', '收支金额': 'amount', '交易金额': 'amount',
+        '类型': 'type', '收支类型': 'type', '交易类型': 'type',
+        '类别': 'category', '分类': 'category', '收支类别': 'category',
+        '部门': 'department', '所属部门': 'department'
+    }
+    
+    df.columns = [column_mapping.get(col, col) for col in df.columns]
+    
+    # 确保必要列存在
+    required_columns = ['date', 'amount', 'type']
+    for col in required_columns:
+        if col not in df.columns:
+            raise Exception(f"缺少必要列: {col}")
+    
+    # 日期处理
+    df['date'] = pd.to_datetime(df['date'], errors='coerce')
+    df = df.dropna(subset=['date'])
+    
+    # 数值处理
+    df['amount'] = pd.to_numeric(df['amount'], errors='coerce')
+    df['amount'] = df['amount'].fillna(0)
+    
+    # 分类字段处理
+    if 'category' not in df.columns:
+        df['category'] = '未分类'
+    else:
+        df['category'] = df['category'].fillna('未分类')
+    
+    if 'department' not in df.columns:
+        df['department'] = '通用部门'
+    else:
+        df['department'] = df['department'].fillna('通用部门')
+    
+    # 添加时间维度
+    df['year'] = df['date'].dt.year
+    df['month'] = df['date'].dt.month
+    df['quarter'] = df['date'].dt.quarter
+    df['year_month'] = df['date'].dt.to_period('M')
+    df['day_of_week'] = df['date'].dt.day_name()
+    
+    return df
+
+def apply_filters(df: pd.DataFrame, period: Optional[str], 
+                 department: Optional[str]) -> pd.DataFrame:
+    """应用时间和部门筛选"""
+    if period:
+        if period == 'last_12_months':
+            cutoff_date = datetime.now() - timedelta(days=365)
+            df = df[df['date'] >= cutoff_date]
+        elif ':' in period:
+            start_str, end_str = period.split(':')
+            start_date = pd.to_datetime(start_str)
+            end_date = pd.to_datetime(end_str)
+            df = df[(df['date'] >= start_date) & (df['date'] <= end_date)]
+    
+    if department:
+        df = df[df['department'] == department]
+    
+    return df
+
+def generate_financial_overview(df: pd.DataFrame) -> str:
+    """生成财务概览报告"""
+    total_income = df[df['type'] == '收入']['amount'].sum()
+    total_expense = df[df['type'] == '支出']['amount'].sum()
+    net_profit = total_income - total_expense
+    
+    # 基本统计
+    total_transactions = len(df)
+    avg_transaction_amount = df['amount'].mean()
+    
+    # 时间范围
+    date_range = f"{df['date'].min().strftime('%Y-%m-%d')} 至 {df['date'].max().strftime('%Y-%m-%d')}"
+    
+    # 部门统计
+    dept_stats = df.groupby('department').agg({
+        'amount': ['sum', 'count']
+    }).round(2)
+    
+    overview = f"""
+🏢 **公司财务概览报告**
+**分析时间范围:** {date_range}
+
+📊 **核心财务指标:**
+• 总收入: ¥{total_income:,.2f}
+• 总支出: ¥{total_expense:,.2f}
+• 净利润: ¥{net_profit:,.2f}
+• 利润率: {net_profit/total_income*100 if total_income > 0 else 0:.1f}%
+
+📈 **交易统计:**
+• 总交易笔数: {total_transactions}笔
+• 平均交易金额: ¥{avg_transaction_amount:,.2f}
+• 收入交易: {len(df[df['type'] == '收入'])}笔
+• 支出交易: {len(df[df['type'] == '支出'])}笔
+
+👥 **部门贡献概览:**
+{dept_stats.to_string()}
+"""
+    
+    return overview
+
+def analyze_financial_trends(df: pd.DataFrame) -> str:
+    """分析财务趋势"""
+    # 月度趋势分析
+    monthly_data = df.groupby('year_month').agg({
+        'amount': ['sum', 'count', 'mean']
+    }).round(2)
+    
+    # 收入支出对比
+    monthly_breakdown = df.groupby(['year_month', 'type'])['amount'].sum().unstack(fill_value=0)
+    monthly_breakdown['净利润'] = monthly_breakdown.get('收入', 0) - monthly_breakdown.get('支出', 0)
+    
+    trend_analysis = f"""
+📈 **财务趋势分析报告**
+
+📅 **月度趋势指标:**
+{monthly_data.to_string()}
+
+💰 **收入支出对比:**
+{monthly_breakdown.to_string()}
+
+🔍 **趋势洞察:**
+• 月均交易额: ¥{monthly_data[('amount', 'sum')].mean():,.2f}
+• 月均交易笔数: {monthly_data[('amount', 'count')].mean():.0f}
+• 增长趋势: {'上升' if monthly_breakdown['净利润'].iloc[-1] > monthly_breakdown['净利润'].iloc[0] else '平稳'}
+"""
+    
+    return trend_analysis
+
+def analyze_category_performance(df: pd.DataFrame) -> str:
+    """分析收支类别绩效"""
+    category_stats = df.groupby(['category', 'type']).agg({
+        'amount': ['sum', 'count', 'mean']
+    }).round(2)
+    
+    category_stats.columns = ['总金额', '交易笔数', '平均金额']
+    category_stats = category_stats.reset_index()
+    
+    # Top类别分析
+    top_income = category_stats[category_stats['type'] == '收入'].nlargest(5, '总金额')
+    top_expense = category_stats[category_stats['type'] == '支出'].nlargest(5, '总金额')
+    
+    category_analysis = f"""
+🏷️ **收支类别绩效分析**
+
+📊 **Top 5 收入类别:**
+{top_income[['category', '总金额', '交易笔数']].to_string(index=False)}
+
+📊 **Top 5 支出类别:**
+{top_expense[['category', '总金额', '交易笔数']].to_string(index=False)}
+
+💡 **效率指标:**
+• 收入集中度: 前3类别占比 {top_income['总金额'].head(3).sum()/category_stats[category_stats['type']=='收入']['总金额'].sum()*100:.1f}%
+• 支出集中度: 前3类别占比 {top_expense['总金额'].head(3).sum()/category_stats[category_stats['type']=='支出']['总金额'].sum()*100:.1f}%
+"""
+    
+    return category_analysis
+
+def analyze_budget_variance(df: pd.DataFrame) -> str:
+    """分析预算执行情况"""
+    # 这里使用模拟预算数据，实际应用中应该从数据库或配置文件中获取
+    budget_data = {
+        '销售收入': 1000000,
+        '技术服务': 500000,
+        '人力成本': 400000,
+        '营销费用': 200000,
+        '研发支出': 300000,
+        '行政管理': 150000
+    }
+    
+    variance_report = "💰 **预算执行分析报告**\n\n"
+    
+    actual_income = df[df['type'] == '收入'].groupby('category')['amount'].sum()
+    actual_expense = df[df['type'] == '支出'].groupby('category')['amount'].sum()
+    
+    for category, budget in budget_data.items():
+        if category in actual_income.index:
+            actual = actual_income[category]
+            variance = (actual - budget) / budget * 100
+            status = "✅ 超额完成" if variance >= 0 else "⚠️ 未达预算"
+            variance_report += f"• {category}: 实际¥{actual:,.0f} / 预算¥{budget:,.0f} ({variance:+.1f}%) {status}\n"
+        elif category in actual_expense.index:
+            actual = actual_expense[category]
+            variance = (actual - budget) / budget * 100
+            status = "⚠️ 超预算" if variance > 0 else "✅ 预算内"
+            variance_report += f"• {category}: 实际¥{actual:,.0f} / 预算¥{budget:,.0f} ({variance:+.1f}%) {status}\n"
+    
+    return variance_report
+
+def analyze_cash_flow(df: pd.DataFrame) -> str:
+    """分析现金流状况"""
+    monthly_cash_flow = df.groupby('year_month').apply(
+        lambda x: x[x['type'] == '收入']['amount'].sum() - 
+                 x[x['type'] == '支出']['amount'].sum()
+    )
+    
+    cash_flow_analysis = f"""
+💳 **现金流分析报告**
+
+📊 **现金流指标:**
+• 月均现金流: ¥{monthly_cash_flow.mean():,.2f}
+• 现金流波动率: {monthly_cash_flow.std():.2f}
+• 正现金流月份: {(monthly_cash_flow > 0).sum()}个月
+• 负现金流月份: {(monthly_cash_flow < 0).sum()}个月
+
+🔔 **现金流健康状况:**
+{generate_cash_flow_health_assessment(monthly_cash_flow)}
+"""
+    
+    return cash_flow_analysis
+
+def analyze_department_performance(df: pd.DataFrame) -> str:
+    """部门绩效分析"""
+    dept_performance = df.groupby('department').agg({
+        'amount': ['sum', 'count', 'mean']
+    }).round(2)
+    
+    dept_performance.columns = ['总金额', '交易笔数', '平均金额']
+    
+    # 部门贡献分析
+    dept_income = df[df['type'] == '收入'].groupby('department')['amount'].sum()
+    dept_expense = df[df['type'] == '支出'].groupby('department')['amount'].sum()
+    
+    dept_analysis = f"""
+👥 **部门绩效分析报告**
+
+📊 **各部门财务表现:**
+{dept_performance.to_string()}
+
+🎯 **部门贡献分析:**
+• 收入Top部门: {dept_income.nlargest(3).to_string()}
+• 支出Top部门: {dept_expense.nlargest(3).to_string()}
+• 净收益最佳部门: {(dept_income - dept_expense).nlargest(3).to_string()}
+"""
+    
+    return dept_analysis
+
+def generate_financial_charts(df: pd.DataFrame, analysis_type: str) -> str:
+    """生成财务分析图表"""
+    plt.figure(figsize=(12, 8))
+    
+    if analysis_type == AnalysisType.OVERVIEW.value:
+        # 收支趋势图
+        monthly_data = df.groupby(['year_month', 'type'])['amount'].sum().unstack()
+        monthly_data.plot(kind='line', ax=plt.gca())
+        plt.title('月度收支趋势')
+        plt.xticks(rotation=45)
+        
+    elif analysis_type == AnalysisType.CATEGORY.value:
+        # 类别分布图
+        category_totals = df.groupby('category')['amount'].sum().nlargest(8)
+        plt.pie(category_totals.values, labels=category_totals.index, autopct='%1.1f%%')
+        plt.title('消费类别分布')
+    
+    plt.tight_layout()
+    
+    # 转换为base64编码图像
+    img_buffer = io.BytesIO()
+    plt.savefig(img_buffer, format='png', bbox_inches='tight')
+    img_buffer.seek(0)
+    img_str = base64.b64encode(img_buffer.read()).decode()
+    
+    return f"data:image/png;base64,{img_str}"
+
+def generate_detailed_report(df: pd.DataFrame, analysis_type: str) -> str:
+    """生成详细分析报告"""
+    base_report = ""
+    if analysis_type == AnalysisType.OVERVIEW.value:
+        base_report = generate_financial_overview(df)
+    # 其他类型的详细报告...
+    
+    detailed_stats = f"""
+📋 **详细统计数据:**
+
+📈 **描述性统计:**
+{df['amount'].describe().to_string()}
+
+📅 **时间范围统计:**
+• 最早交易: {df['date'].min().strftime('%Y-%m-%d')}
+• 最晚交易: {df['date'].max().strftime('%Y-%m-%d')}
+• 分析天数: {(df['date'].max() - df['date'].min()).days}天
+"""
+    
+    return base_report + detailed_stats
+
+def generate_cash_flow_health_assessment(monthly_cash_flow: pd.Series) -> str:
+    """生成现金流健康评估"""
+    negative_months = (monthly_cash_flow < 0).sum()
+    total_months = len(monthly_cash_flow)
+    
+    if negative_months == 0:
+        return "🟢 优秀: 所有月份均为正现金流，财务状况非常健康"
+    elif negative_months <= total_months * 0.25:
+        return "🟡 良好: 少数月份出现负现金流，整体状况良好"
+    elif negative_months <= total_months * 0.5:
+        return "🟠 一般: 较多月份出现负现金流，需要关注资金管理"
+    else:
+        return "🔴 风险: 大部分月份为负现金流，存在资金风险"
+
+# 快速财务检查工具
+@tool
+def quick_financial_check(file_path: str) -> str:
+    """快速财务健康检查工具"""
+    try:
+        df = load_financial_data(file_path)
+        df = preprocess_financial_data(df)
+        
+        overview = generate_financial_overview(df)
+        cash_flow = analyze_cash_flow(df)
+        
+        return f"{overview}\n\n{cash_flow}"
+    except Exception as e:
+        return f"快速检查失败: {str(e)}"
+
+# 初始化其他工具
+final_answer = FinalAnswerTool()
+
+# 配置模型
+model = HfApiModel(
+    max_tokens=2096,
+    temperature=0.5,
+    model_id='Qwen/Qwen2.5-Coder-32B-Instruct',
+    custom_role_conversions=None,
+)
+
+# 加载其他工具
+image_generation_tool = load_tool("agents-course/text-to-image", trust_remote_code=True)
+
+# 加载提示模板
+with open("prompts.yaml", 'r') as stream:
+    prompt_templates = yaml.safe_load(stream)
+
+# 创建代理并集成所有工具
+agent = CodeAgent(
+    model=model,
+    tools=[
+        final_answer,
+        analyze_corporate_finance,  # 公司财务分析工具
+        quick_financial_check,      # 快速检查工具
+        image_generation_tool,
+        get_current_time_in_timezone,
+        my_custom_tool
+    ],
+    max_steps=12,  # 增加步骤限制以支持复杂分析
+    verbosity_level=1,
+    grammar=None,
+    planning_interval=None,
+    name="Corporate Finance Analyst Pro",
+    description="专业的企业财务分析AI助手，支持多维度收支数据分析和可视化",
+    prompt_templates=prompt_templates
+)
+
+# 启动Gradio界面
+GradioUI(agent).launch()

app.py

@@ -1,69 +0,0 @@
-from smolagents import CodeAgent,DuckDuckGoSearchTool, HfApiModel,load_tool,tool
-import datetime
-import requests
-import pytz
-import yaml
-from tools.final_answer import FinalAnswerTool
-
-from Gradio_UI import GradioUI
-
-# Below is an example of a tool that does nothing. Amaze us with your creativity !
-@tool
-def my_custom_tool(arg1:str, arg2:int)-> str: #it's import to specify the return type
-    #Keep this format for the description / args / args description but feel free to modify the tool
-    """A tool that does nothing yet 
-    Args:
-        arg1: the first argument
-        arg2: the second argument
-    """
-    return "What magic will you build ?"
-
-@tool
-def get_current_time_in_timezone(timezone: str) -> str:
-    """A tool that fetches the current local time in a specified timezone.
-    Args:
-        timezone: A string representing a valid timezone (e.g., 'America/New_York').
-    """
-    try:
-        # Create timezone object
-        tz = pytz.timezone(timezone)
-        # Get current time in that timezone
-        local_time = datetime.datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S")
-        return f"The current local time in {timezone} is: {local_time}"
-    except Exception as e:
-        return f"Error fetching time for timezone '{timezone}': {str(e)}"
-
-
-final_answer = FinalAnswerTool()
-
-# If the agent does not answer, the model is overloaded, please use another model or the following Hugging Face Endpoint that also contains qwen2.5 coder:
-# model_id='https://pflgm2locj2t89co.us-east-1.aws.endpoints.huggingface.cloud' 
-
-model = HfApiModel(
-max_tokens=2096,
-temperature=0.5,
-model_id='Qwen/Qwen2.5-Coder-32B-Instruct',# it is possible that this model may be overloaded
-custom_role_conversions=None,
-)
-
-
-# Import tool from Hub
-image_generation_tool = load_tool("agents-course/text-to-image", trust_remote_code=True)
-
-with open("prompts.yaml", 'r') as stream:
-    prompt_templates = yaml.safe_load(stream)
-    
-agent = CodeAgent(
-    model=model,
-    tools=[final_answer], ## add your tools here (don't remove final answer)
-    max_steps=6,
-    verbosity_level=1,
-    grammar=None,
-    planning_interval=None,
-    name=None,
-    description=None,
-    prompt_templates=prompt_templates
-)
-
-
-GradioUI(agent).launch()