AI-video-lifeplan-demo/backend/data_builder.py

"""
Chart data builder and financial simulation engine

This module:
- Simulates future asset projections
- Handles inflation calculations
- Calculates compound interest on deficits
- Generates data in format compatible with Remotion charts
"""

from typing import Dict, Any, List
import math


class FinancialSimulator:
    """Simulate financial projections for Life Plan analysis"""
    
    def __init__(self, config: Optional[Dict[str, Any]] = None):
        self.config = config or {}
        self.inflation_rate = self.config.get("inflation_rate", 0.02)  # 2% default
        self.deficit_interest_rate = self.config.get("deficit_interest_rate", 0.15)  # 15% default
        self.investment_return = self.config.get("investment_return", 0.03)  # 3% default
    
    def simulate_timeline(
        self,
        current_age: int,
        retirement_age: int,
        life_expectancy: int,
        current_assets: float,
        annual_income: float,
        annual_expenses: float,
        life_events: Optional[List[Dict]] = None,
    ) -> List[Dict[str, Any]]:
        """
        Simulate year-by-year financial timeline
        
        Args:
            current_age: Current age of representative
            retirement_age: Planned retirement age
            life_expectancy: Expected life expectancy
            current_assets: Current asset balance (万円)
            annual_income: Annual income (万円)
            annual_expenses: Annual expenses (万円)
            life_events: List of major life events {year, type, amount}
            
        Returns:
            List of yearly snapshots compatible with YearlySnapshot interface
        """
        timeline = []
        current_year = 2025  # TODO: Make this dynamic
        asset_balance = current_assets
        
        life_events = life_events or []
        
        for age in range(current_age, life_expectancy + 1):
            year = current_year + (age - current_age)
            
            # Adjust income (stops at retirement)
            income = annual_income if age < retirement_age else 0
            
            # Apply inflation to expenses
            years_from_now = age - current_age
            inflated_expenses = annual_expenses * math.pow(1 + self.inflation_rate, years_from_now)
            
            # Check for life events
            event_cost = self._get_event_cost(year, life_events)
            total_expenses = inflated_expenses + event_cost
            
            # Calculate net cash flow
            net_cash_flow = income - total_expenses
            
            # Update asset balance
            if asset_balance >= 0:
                # Positive balance: apply investment return
                asset_balance = asset_balance * (1 + self.investment_return) + net_cash_flow
            else:
                # Negative balance: apply deficit interest (compound debt)
                asset_balance = asset_balance * (1 + self.deficit_interest_rate) + net_cash_flow
            
            # Create snapshot
            snapshot = {
                "year": year,
                "ages": {
                    "rep": age,
                    "spouse": None,  # TODO: Add spouse calculation
                },
                "financials": {
                    "income": round(income, 0),
                    "expenditure": round(total_expenses, 0),
                    "assetBalance": round(asset_balance, 0),
                },
                "events": self._get_event_labels(year, life_events),
                "isCritical": asset_balance < 0,
            }
            
            timeline.append(snapshot)
        
        return timeline
    
    def _get_event_cost(self, year: int, life_events: List[Dict]) -> float:
        """Get total event cost for a specific year"""
        total = 0
        for event in life_events:
            if event.get("year") == year:
                total += event.get("amount", 0)
        return total
    
    def _get_event_labels(self, year: int, life_events: List[Dict]) -> List[str]:
        """Get event labels for a specific year"""
        labels = []
        for event in life_events:
            if event.get("year") == year:
                labels.append(event.get("label", "Event"))
        return labels
    
    def build_cashflow_data(self, timeline: List[Dict]) -> List[Dict]:
        """
        Convert timeline to CashFlow chart format
        
        Args:
            timeline: YearlySnapshot format timeline
            
        Returns:
            List compatible with CashFlowData interface
        """
        cashflow = []
        
        for snapshot in timeline:
            cashflow.append({
                "year": snapshot["year"],
                "age": snapshot["ages"]["rep"],
                "income": {
                    "total": snapshot["financials"]["income"],
                },
                "expenses": {
                    "living": snapshot["financials"]["expenditure"] * 0.4,  # Mock breakdown
                    "housing": snapshot["financials"]["expenditure"] * 0.3,
                    "education": snapshot["financials"]["expenditure"] * 0.15,
                    "tax": snapshot["financials"]["expenditure"] * 0.1,
                    "event": snapshot["financials"]["expenditure"] * 0.05,
                },
                "totalExpense": snapshot["financials"]["expenditure"],
                "netCashFlow": snapshot["financials"]["income"] - snapshot["financials"]["expenditure"],
                "assetBalance": snapshot["financials"]["assetBalance"],
            })
        
        return cashflow


def build_chart_data(pdf_data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Build chart data from PDF extraction results
    
    Args:
        pdf_data: Structured data from PDF extraction
        
    Returns:
        Dictionary with 'assetDepletion' and 'cashFlow' chart data
    """
    simulator = FinancialSimulator()
    
    # Extract parameters from PDF data
    current_age = pdf_data.get("current_age", 35)
    retirement_age = pdf_data.get("retirement_age", 65)
    life_expectancy = pdf_data.get("life_expectancy", 100)
    current_assets = pdf_data.get("current_assets", 5000)
    annual_income = pdf_data.get("annual_income", 600)
    annual_expenses = pdf_data.get("annual_expenses", 450)
    life_events = pdf_data.get("life_events", [])
    
    # Simulate timeline
    timeline = simulator.simulate_timeline(
        current_age=current_age,
        retirement_age=retirement_age,
        life_expectancy=life_expectancy,
        current_assets=current_assets,
        annual_income=annual_income,
        annual_expenses=annual_expenses,
        life_events=life_events,
    )
    
    # Build cashflow data
    cashflow = simulator.build_cashflow_data(timeline)
    
    return {
        "assetDepletion": {
            "timeline": timeline,
            "meta": {
                "currencyUnit": "万円",
                "startYear": timeline[0]["year"] if timeline else 2025,
            },
        },
        "cashFlow": {
            "timeline": cashflow,
            "meta": {
                "currencyUnit": "万円",
            },
        },
    }