"""Comprehensive Financial Analysis for FY25 YTD using FCCS MCP Agent.
This script generates:
1. Top 10 entities by revenue
2. Bottom 5 entities by profit margin
3. Largest variances vs FY24
4. Intercompany matching status
"""
import asyncio
import sys
from pathlib import Path
from typing import Optional, Dict, List
from datetime import datetime
# Add parent directory to path
sys.path.insert(0, str(Path(__file__).parent))
from fccs_agent.config import load_config
from fccs_agent.agent import initialize_agent, close_agent
from fccs_agent.tools.data import smart_retrieve
from fccs_agent.utils.cache import load_members_from_cache
async def get_account_value(
account: str,
entity: str,
year: str,
period: str = "Dec"
) -> Optional[float]:
"""Get account value for an entity using smart_retrieve."""
try:
result = await smart_retrieve(
account=account,
entity=entity,
period=period,
years=year,
scenario="Actual"
)
if result.get("status") == "success":
data = result.get("data", {})
rows = data.get("rows", [])
if rows and rows[0].get("data"):
value = rows[0]["data"][0]
return float(value) if value is not None else None
except Exception:
pass
return None
async def get_entity_metrics(entity: str) -> Dict:
"""Get revenue, net income, and profit margin for FY24 and FY25."""
metrics = {
"entity": entity,
"fy24": {"revenue": None, "net_income": None, "profit_margin": None},
"fy25": {"revenue": None, "net_income": None, "profit_margin": None},
"variance": {"revenue": None, "net_income": None, "revenue_pct": None, "net_income_pct": None}
}
# Get FY24 data
fy24_revenue = await get_account_value("FCCS_Sales", entity, "FY24", "Dec")
fy24_net_income = await get_account_value("FCCS_Net Income", entity, "FY24", "Dec")
metrics["fy24"]["revenue"] = fy24_revenue
metrics["fy24"]["net_income"] = fy24_net_income
if fy24_revenue and fy24_revenue != 0:
metrics["fy24"]["profit_margin"] = (fy24_net_income / fy24_revenue * 100) if fy24_net_income else None
# Get FY25 data
fy25_revenue = await get_account_value("FCCS_Sales", entity, "FY25", "Dec")
fy25_net_income = await get_account_value("FCCS_Net Income", entity, "FY25", "Dec")
metrics["fy25"]["revenue"] = fy25_revenue
metrics["fy25"]["net_income"] = fy25_net_income
if fy25_revenue and fy25_revenue != 0:
metrics["fy25"]["profit_margin"] = (fy25_net_income / fy25_revenue * 100) if fy25_net_income else None
# Calculate variances
if fy24_revenue is not None and fy25_revenue is not None:
metrics["variance"]["revenue"] = fy25_revenue - fy24_revenue
if fy24_revenue != 0:
metrics["variance"]["revenue_pct"] = (metrics["variance"]["revenue"] / abs(fy24_revenue)) * 100
if fy24_net_income is not None and fy25_net_income is not None:
metrics["variance"]["net_income"] = fy25_net_income - fy24_net_income
if fy24_net_income != 0:
metrics["variance"]["net_income_pct"] = (metrics["variance"]["net_income"] / abs(fy24_net_income)) * 100
return metrics
async def generate_comprehensive_analysis():
"""Generate comprehensive financial analysis for FY25 YTD."""
print("=" * 100)
print("COMPREHENSIVE FINANCIAL ANALYSIS - FY25 YTD")
print("=" * 100)
print(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print()
try:
# Initialize agent
config = load_config()
app_name = await initialize_agent(config)
print(f"[OK] Connected to FCCS: {app_name}")
print()
# Load entity list from cache
print("Loading entity list...")
cached_entities = load_members_from_cache("Consol", "Entity")
entities = [item.get("name") for item in cached_entities.get("items", []) if item.get("name")]
# Filter out totals and system entities
exclude_keywords = ["Total", "FCCS_Total", "FCCS_Entity Total", "FCCS_Global Assumptions",
"Elim", "Segment", "Consolidated", "Consolidation"]
individual_entities = [
e for e in entities
if not any(kw in e for kw in exclude_keywords) and e not in ["Root"]
]
print(f"[OK] Found {len(individual_entities)} individual entities")
print()
# Collect metrics for all entities
print("Collecting financial metrics for all entities...")
print("(This may take several minutes...)")
print()
all_metrics = []
processed = 0
for entity in individual_entities:
if processed % 50 == 0:
print(f" Processed {processed}/{len(individual_entities)} entities...")
metrics = await get_entity_metrics(entity)
# Only include entities with FY25 revenue data
if metrics["fy25"]["revenue"] is not None:
all_metrics.append(metrics)
processed += 1
print(f"[OK] Collected metrics for {len(all_metrics)} entities with FY25 data")
print()
# 1. TOP 10 ENTITIES BY REVENUE
print("=" * 100)
print("1. TOP 10 ENTITIES BY REVENUE (FY25 YTD)")
print("=" * 100)
print()
top_revenue = sorted(
all_metrics,
key=lambda x: x["fy25"]["revenue"] if x["fy25"]["revenue"] else float('-inf'),
reverse=True
)[:10]
print(f"{'Rank':<6} {'Entity':<40} {'Revenue (FY25)':>20} {'Net Income':>20} {'Margin %':>12}")
print("-" * 100)
for rank, metrics in enumerate(top_revenue, 1):
entity = metrics["entity"][:38] # Truncate if too long
revenue = metrics["fy25"]["revenue"] or 0
net_income = metrics["fy25"]["net_income"] or 0
margin = metrics["fy25"]["profit_margin"] or 0
print(f"{rank:<6} {entity:<40} ${revenue:>19,.2f} ${net_income:>19,.2f} {margin:>11.2f}%")
print()
# 2. BOTTOM 5 ENTITIES BY PROFIT MARGIN
print("=" * 100)
print("2. BOTTOM 5 ENTITIES BY PROFIT MARGIN (FY25 YTD)")
print("=" * 100)
print()
print("(Entities with revenue > 0 and lowest profit margins)")
print()
# Filter entities with revenue > 0
entities_with_revenue = [
m for m in all_metrics
if m["fy25"]["revenue"] and m["fy25"]["revenue"] > 0
]
bottom_margin = sorted(
entities_with_revenue,
key=lambda x: x["fy25"]["profit_margin"] if x["fy25"]["profit_margin"] is not None else float('inf')
)[:5]
print(f"{'Rank':<6} {'Entity':<40} {'Revenue':>20} {'Net Income':>20} {'Margin %':>12}")
print("-" * 100)
for rank, metrics in enumerate(bottom_margin, 1):
entity = metrics["entity"][:38]
revenue = metrics["fy25"]["revenue"] or 0
net_income = metrics["fy25"]["net_income"] or 0
margin = metrics["fy25"]["profit_margin"] or 0
print(f"{rank:<6} {entity:<40} ${revenue:>19,.2f} ${net_income:>19,.2f} {margin:>11.2f}%")
print()
# 3. LARGEST VARIANCES VS FY24
print("=" * 100)
print("3. LARGEST VARIANCES VS FY24")
print("=" * 100)
print()
# Filter entities with both FY24 and FY25 data
entities_with_variance = [
m for m in all_metrics
if m["fy24"]["revenue"] is not None and m["fy25"]["revenue"] is not None
]
# Largest revenue variances (absolute)
print("3a. TOP 10 LARGEST REVENUE VARIANCES (Absolute)")
print("-" * 100)
print()
revenue_variances = sorted(
entities_with_variance,
key=lambda x: abs(x["variance"]["revenue"]) if x["variance"]["revenue"] is not None else 0,
reverse=True
)[:10]
print(f"{'Rank':<6} {'Entity':<40} {'FY24 Revenue':>20} {'FY25 Revenue':>20} {'Variance':>20} {'% Change':>12}")
print("-" * 100)
for rank, metrics in enumerate(revenue_variances, 1):
entity = metrics["entity"][:38]
fy24_rev = metrics["fy24"]["revenue"] or 0
fy25_rev = metrics["fy25"]["revenue"] or 0
variance = metrics["variance"]["revenue"] or 0
pct_change = metrics["variance"]["revenue_pct"] or 0
print(f"{rank:<6} {entity:<40} ${fy24_rev:>19,.2f} ${fy25_rev:>19,.2f} ${variance:>19,.2f} {pct_change:>11.2f}%")
print()
print("3b. TOP 10 LARGEST NET INCOME VARIANCES (Absolute)")
print("-" * 100)
print()
net_income_variances = sorted(
entities_with_variance,
key=lambda x: abs(x["variance"]["net_income"]) if x["variance"]["net_income"] is not None else 0,
reverse=True
)[:10]
print(f"{'Rank':<6} {'Entity':<40} {'FY24 Net Inc':>20} {'FY25 Net Inc':>20} {'Variance':>20} {'% Change':>12}")
print("-" * 100)
for rank, metrics in enumerate(net_income_variances, 1):
entity = metrics["entity"][:38]
fy24_ni = metrics["fy24"]["net_income"] or 0
fy25_ni = metrics["fy25"]["net_income"] or 0
variance = metrics["variance"]["net_income"] or 0
pct_change = metrics["variance"]["net_income_pct"] or 0
print(f"{rank:<6} {entity:<40} ${fy24_ni:>19,.2f} ${fy25_ni:>19,.2f} ${variance:>19,.2f} {pct_change:>11.2f}%")
print()
# 4. INTERCOMPANY MATCHING STATUS
print("=" * 100)
print("4. INTERCOMPANY MATCHING STATUS")
print("=" * 100)
print()
print("Attempting to generate intercompany matching report...")
try:
# Use MCP tool for intercompany matching report
from fccs_agent.tools.consolidation import generate_intercompany_matching_report
ic_result = await generate_intercompany_matching_report(
parameters={"scenario": "Actual", "year": "FY25", "period": "Dec"}
)
print("[OK] Intercompany matching report generated successfully")
if isinstance(ic_result, dict):
if ic_result.get("status") == "success":
print(f"Report details: {ic_result.get('data', 'N/A')}")
else:
print(f"Report result: {ic_result}")
else:
print(f"Report result: {ic_result}")
except Exception as e:
print(f"[WARNING] Could not generate intercompany matching report: {str(e)}")
print(" This may require additional permissions or the report may need to be run manually.")
print(" You can generate this report manually in FCCS under Intercompany Matching.")
print()
# SUMMARY STATISTICS
print("=" * 100)
print("SUMMARY STATISTICS")
print("=" * 100)
print()
total_fy25_revenue = sum(m["fy25"]["revenue"] or 0 for m in all_metrics)
total_fy25_net_income = sum(m["fy25"]["net_income"] or 0 for m in all_metrics)
total_fy24_revenue = sum(m["fy24"]["revenue"] or 0 for m in entities_with_variance)
total_fy24_net_income = sum(m["fy24"]["net_income"] or 0 for m in entities_with_variance)
print(f"Total Entities Analyzed: {len(all_metrics)}")
print(f"Entities with FY24 Comparison Data: {len(entities_with_variance)}")
print()
print(f"Total FY25 YTD Revenue: ${total_fy25_revenue:>20,.2f}")
print(f"Total FY25 YTD Net Income: ${total_fy25_net_income:>20,.2f}")
if total_fy25_revenue > 0:
overall_margin = (total_fy25_net_income / total_fy25_revenue) * 100
print(f"Overall Profit Margin: {overall_margin:>20.2f}%")
print()
if total_fy24_revenue > 0:
revenue_variance_total = total_fy25_revenue - total_fy24_revenue
revenue_variance_pct = (revenue_variance_total / total_fy24_revenue) * 100
print(f"Total FY24 YTD Revenue: ${total_fy24_revenue:>20,.2f}")
print(f"Revenue Variance: ${revenue_variance_total:>20,.2f} ({revenue_variance_pct:+.2f}%)")
print()
if total_fy24_net_income != 0:
net_income_variance_total = total_fy25_net_income - total_fy24_net_income
net_income_variance_pct = (net_income_variance_total / abs(total_fy24_net_income)) * 100
print(f"Total FY24 YTD Net Income: ${total_fy24_net_income:>20,.2f}")
print(f"Net Income Variance: ${net_income_variance_total:>20,.2f} ({net_income_variance_pct:+.2f}%)")
print()
print("=" * 100)
print("ANALYSIS COMPLETE")
print("=" * 100)
# Close agent
await close_agent()
except Exception as e:
print(f"[ERROR] {str(e)}")
import traceback
traceback.print_exc()
try:
await close_agent()
except:
pass
if __name__ == "__main__":
asyncio.run(generate_comprehensive_analysis())
