Data Storyteller

Bud1  
tslg1Scomp

 
@
�





 
@
�






 
@
�






 
@assetslg1ScompassetsmoDDblob�a#�w�AassetsmodDblob�a#�w�Aassetsph1Scomp
referenceslg1Scomp
referencesmoDDblob��a#�w�A
referencesmodDblob��a#�w�A
referencesph1Scompscriptsbwspblob�bplist00�
]ShowStatusBar[ShowToolbar[ShowTabView_ContainerShowSidebar\WindowBounds[ShowSidebar				_{{233, 189}, {920, 436}}	#/;R_klmno�


�scriptslg1Scomp��scriptslsvCblobbbplist00�
	
XYZ[\]Z_useRelativeDates_showIconPreviewWcolumns_calculateAllSizes_viewOptionsVersion_scrollPositionYXtextSize_scrollPositionXZsortColumnXiconSizeZaxTextSize		� %).38=AFJNR�WvisibleUwidthYascendingZidentifier	
,	Tname�UwidthYascendingWvisibleXubiquity#�"$	�\dateModified�"([dateCreated�+-	aTsize�02	s	Tkind�57d	Ulabel�:<K	Wversion�@	Xcomments�CE�^dateLastOpened�CIZshareOwner�CM_shareLastEditor�O"YdateAdded�TV�_invitationStatus
##@.#�c�Tname#@02DL`u���������������





!
)
2
4
5
6
?
@
B
C
P
Y
Z
[
g
p
q
s
t
y
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�)*+4578KL]fox}
^�scriptslsvpblob�bplist00�
	
 HIJKLMJ_useRelativeDates_showIconPreviewWcolumns_calculateAllSizes_viewOptionsVersion_scrollPositionYXtextSize_scrollPositionXZsortColumnXiconSizeZaxTextSize		�!&+059>BXcommentsUlabelWversion[dateCreatedTsize\dateModifiedTkindTname^dateLastOpened� UindexUwidthYascendingWvisible
,	�"# d	�'( K	�,-  ��12 a	�6- 
	�:;s		�?		�CD  �
##@.#�c�Tname#@02DL`u��������������



$
-
3
9
C
K
M
P
Q
R
[
]
_
`
a
j
l
n
o
p
y
{
}
~

�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
N
scriptsmoDDblobV�u�w�AscriptsmodDblobV�u�w�Ascriptsph1Scomp�scriptsvSrnlong
 E
DSDB
 `
�



(
0
@
�






 
@
�






 
@��





!
)
2
4
5
6
?
@
B
C
P
Y
Z
[
g
p
q
s
t
y
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�
�)*+4578KL]fox}
^�scriptslsvpblob�bplist00�
	
 HIJKLMJ_useRelativeDates_showIconPreviewWcolumns_calculateAllSizes_viewOptionsVersion_scrollPositionYXtextSize_scrollPositionXZsortColumnXiconSizeZaxTextSize		�!&+059>BXcommentsUlabelWversion[dateCreatedTsize\dateModifiedTkindTname^dateLastOpened� UindexUwidthYascendingWvisible
,	�"# d	�'( K	�,-  ��12 a	�6- 
	�:;s		�?		�CD  �
##@.#�c�Tname#@02DL`u��������������




display_name: 'Data Storyteller'
short_description: 'Analyze datasets and produce concise narrative findings.'
default_prompt: 'Help me analyze this dataset and summarize the findings.'

#!/usr/bin/env python3
"""
A/B Test Calculator - Statistical significance testing for A/B experiments.
"""

import argparse
import json
import math
from typing import Dict, List, Tuple, Optional

import numpy as np
from scipy import stats
from scipy.stats import beta as beta_dist


class ABTestCalculator:
    """Calculate statistical significance for A/B tests."""

    def __init__(self, alpha: float = 0.05):
        """
        Initialize calculator.

        Args:
            alpha: Significance level (default 0.05 for 95% confidence)
        """
        self.alpha = alpha

    def test_significance(self, control_visitors: int, control_conversions: int,
                         variant_visitors: int, variant_conversions: int,
                         test: str = "chi_square") -> Dict:
        """
        Test statistical significance between control and variant.

        Args:
            control_visitors: Number of visitors in control group
            control_conversions: Number of conversions in control group
            variant_visitors: Number of visitors in variant group
            variant_conversions: Number of conversions in variant group
            test: Test method ("chi_square", "z_test")

        Returns:
            Dictionary with test results
        """
        control_rate = control_conversions / control_visitors
        variant_rate = variant_conversions / variant_visitors

        lift = (variant_rate - control_rate) / control_rate if control_rate > 0 else 0
        lift_absolute = variant_rate - control_rate

        if test == "chi_square":
            p_value = self._chi_square_test(
                control_visitors, control_conversions,
                variant_visitors, variant_conversions
            )
        elif test == "z_test":
            p_value = self._z_test_proportions(
                control_visitors, control_conversions,
                variant_visitors, variant_conversions
            )
        else:
            raise ValueError(f"Unknown test method: {test}")

        # Calculate confidence interval for lift
        ci = self._lift_confidence_interval(
            control_visitors, control_conversions,
            variant_visitors, variant_conversions
        )

        significant = p_value < self.alpha

        if significant:
            if lift > 0:
                recommendation = "Variant shows significant improvement. Consider implementing."
            else:
                recommendation = "Variant shows significant decrease. Keep control."
        else:
            recommendation = "No significant difference detected. Need more data or larger effect."

        return {
            "significant": significant,
            "p_value": p_value,
            "control_rate": control_rate,
            "variant_rate": variant_rate,
            "lift": lift,
            "lift_absolute": lift_absolute,
            "confidence_interval": ci,
            "test_method": test,
            "alpha": self.alpha,
            "recommendation": recommendation
        }

    def _chi_square_test(self, c_visitors: int, c_conv: int,
                        v_visitors: int, v_conv: int) -> float:
        """Perform chi-square test for independence."""
        # Contingency table
        observed = np.array([
            [c_conv, c_visitors - c_conv],
            [v_conv, v_visitors - v_conv]
        ])

        chi2, p_value, dof, expected = stats.chi2_contingency(observed)
        return p_value

    def _z_test_proportions(self, c_visitors: int, c_conv: int,
                           v_visitors: int, v_conv: int) -> float:
        """Perform Z-test for two proportions."""
        p1 = c_conv / c_visitors
        p2 = v_conv / v_visitors
        n1 = c_visitors
        n2 = v_visitors

        # Pooled proportion
        p_pooled = (c_conv + v_conv) / (n1 + n2)

        # Standard error
        se = math.sqrt(p_pooled * (1 - p_pooled) * (1/n1 + 1/n2))

        if se == 0:
            return 1.0

        # Z statistic
        z = (p2 - p1) / se

        # Two-tailed p-value
        p_value = 2 * (1 - stats.norm.cdf(abs(z)))
        return p_value

    def _lift_confidence_interval(self, c_visitors: int, c_conv: int,
                                  v_visitors: int, v_conv: int,
                                  confidence: float = 0.95) -> Dict:
        """Calculate confidence interval for relative lift."""
        p1 = c_conv / c_visitors
        p2 = v_conv / v_visitors

        # Standard errors
        se1 = math.sqrt(p1 * (1 - p1) / c_visitors)
        se2 = math.sqrt(p2 * (1 - p2) / v_visitors)

        # SE of difference
        se_diff = math.sqrt(se1**2 + se2**2)

        # Z value for confidence level
        z = stats.norm.ppf((1 + confidence) / 2)

        diff = p2 - p1
        margin = z * se_diff

        # Convert to relative lift if control rate > 0
        if p1 > 0:
            lower = (diff - margin) / p1
            upper = (diff + margin) / p1
        else:
            lower = 0
            upper = 0

        return {
            "lower": lower,
            "upper": upper,
            "confidence_level": confidence
        }

    def calculate_sample_size(self, baseline_rate: float,
                             minimum_detectable_effect: float,
                             power: float = 0.8,
                             alpha: float = None) -> Dict:
        """
        Calculate required sample size per variant.

        Args:
            baseline_rate: Current conversion rate (e.g., 0.05 for 5%)
            minimum_detectable_effect: Relative change to detect (e.g., 0.10 for 10%)
            power: Statistical power (default 0.8 for 80%)
            alpha: Significance level (uses instance default if not specified)

        Returns:
            Dictionary with sample size information
        """
        if alpha is None:
            alpha = self.alpha

        p1 = baseline_rate
        p2 = baseline_rate * (1 + minimum_detectable_effect)

        # Z values
        z_alpha = stats.norm.ppf(1 - alpha/2)
        z_beta = stats.norm.ppf(power)

        # Pooled proportion
        p_pooled = (p1 + p2) / 2

        # Sample size formula
        numerator = (z_alpha * math.sqrt(2 * p_pooled * (1 - p_pooled)) +
                    z_beta * math.sqrt(p1 * (1 - p1) + p2 * (1 - p2))) ** 2
        denominator = (p2 - p1) ** 2

        if denominator == 0:
            sample_size = float('inf')
        else:
            sample_size = math.ceil(numerator / denominator)

        return {
            "sample_size_per_variant": sample_size,
            "total_sample_size": sample_size * 2,
            "baseline_rate": baseline_rate,
            "expected_variant_rate": p2,
            "minimum_detectable_effect": minimum_detectable_effect,
            "power": power,
            "alpha": alpha
        }

    def calculate_power(self, baseline_rate: float,
                       minimum_detectable_effect: float,
                       sample_size: int,
                       alpha: float = None) -> Dict:
        """
        Calculate statistical power given sample size.

        Args:
            baseline_rate: Current conversion rate
            minimum_detectable_effect: Relative change to detect
            sample_size: Sample size per variant
            alpha: Significance level

        Returns:
            Dictionary with power analysis
        """
        if alpha is None:
            alpha = self.alpha

        p1 = baseline_rate
        p2 = baseline_rate * (1 + minimum_detectable_effect)

        # Standard error
        se = math.sqrt(p1 * (1 - p1) / sample_size + p2 * (1 - p2) / sample_size)

        # Z value for alpha
        z_alpha = stats.norm.ppf(1 - alpha/2)

        # Calculate power
        z_beta = (abs(p2 - p1) / se) - z_alpha
        power = stats.norm.cdf(z_beta)

        return {
            "power": power,
            "sample_size_per_variant": sample_size,
            "baseline_rate": baseline_rate,
            "minimum_detectable_effect": minimum_detectable_effect,
            "alpha": alpha,
            "interpretation": f"{power:.0%} chance of detecting the effect if it exists"
        }

    def confidence_interval(self, visitors: int, conversions: int,
                           confidence: float = 0.95) -> Dict:
        """
        Calculate confidence interval for a conversion rate.

        Args:
            visitors: Total visitors
            conversions: Number of conversions
            confidence: Confidence level (default 0.95)

        Returns:
            Dictionary with CI bounds
        """
        rate = conversions / visitors

        # Wilson score interval (better for small samples)
        z = stats.norm.ppf((1 + confidence) / 2)

        denominator = 1 + z**2 / visitors
        center = (rate + z**2 / (2 * visitors)) / denominator
        margin = z * math.sqrt((rate * (1 - rate) + z**2 / (4 * visitors)) / visitors) / denominator

        return {
            "rate": rate,
            "lower": max(0, center - margin),
            "upper": min(1, center + margin),
            "confidence_level": confidence,
            "method": "wilson"
        }

    def bayesian_analysis(self, control_visitors: int, control_conversions: int,
                         variant_visitors: int, variant_conversions: int,
                         simulations: int = 100000) -> Dict:
        """
        Bayesian analysis using Beta distributions.

        Args:
            control_visitors: Number of visitors in control
            control_conversions: Number of conversions in control
            variant_visitors: Number of visitors in variant
            variant_conversions: Number of conversions in variant
            simulations: Number of Monte Carlo simulations

        Returns:
            Dictionary with Bayesian analysis results
        """
        # Beta distribution parameters (using uniform prior)
        alpha_c = control_conversions + 1
        beta_c = control_visitors - control_conversions + 1
        alpha_v = variant_conversions + 1
        beta_v = variant_visitors - variant_conversions + 1

        # Sample from posterior distributions
        control_samples = np.random.beta(alpha_c, beta_c, simulations)
        variant_samples = np.random.beta(alpha_v, beta_v, simulations)

        # Probability variant beats control
        prob_variant_better = np.mean(variant_samples > control_samples)

        # Expected lift
        lift_samples = (variant_samples - control_samples) / control_samples
        expected_lift = np.mean(lift_samples)

        # Credible interval for lift
        ci_lower = np.percentile(lift_samples, 2.5)
        ci_upper = np.percentile(lift_samples, 97.5)

        return {
            "prob_variant_better": prob_variant_better,
            "prob_control_better": 1 - prob_variant_better,
            "expected_lift": expected_lift,
            "credible_interval_95": [ci_lower, ci_upper],
            "control_rate_mean": np.mean(control_samples),
            "variant_rate_mean": np.mean(variant_samples)
        }

    def test_multiple_variants(self, control: Tuple[int, int],
                              variants: List[Tuple[int, int]],
                              correction: str = "bonferroni") -> Dict:
        """
        Test multiple variants against control with correction.

        Args:
            control: (visitors, conversions) for control
            variants: List of (visitors, conversions) tuples for variants
            correction: Multiple testing correction ("bonferroni", "holm", "none")

        Returns:
            Dictionary with all variant comparisons
        """
        c_visitors, c_conversions = control
        c_rate = c_conversions / c_visitors

        n_variants = len(variants)
        results = []
        p_values = []

        # Run tests for each variant
        for i, (v_visitors, v_conversions) in enumerate(variants):
            v_rate = v_conversions / v_visitors
            lift = (v_rate - c_rate) / c_rate if c_rate > 0 else 0

            p_value = self._chi_square_test(
                c_visitors, c_conversions,
                v_visitors, v_conversions
            )
            p_values.append(p_value)

            results.append({
                "name": f"Variant {chr(65 + i)}",
                "visitors": v_visitors,
                "conversions": v_conversions,
                "rate": v_rate,
                "lift": lift,
                "p_value": p_value
            })

        # Apply correction
        if correction == "bonferroni":
            adjusted_alpha = self.alpha / n_variants
            for r in results:
                r["significant"] = r["p_value"] < adjusted_alpha
        elif correction == "holm":
            # Holm-Bonferroni method
            sorted_indices = np.argsort(p_values)
            for rank, idx in enumerate(sorted_indices):
                adjusted_alpha = self.alpha / (n_variants - rank)
                results[idx]["significant"] = p_values[idx] < adjusted_alpha
        else:
            for r in results:
                r["significant"] = r["p_value"] < self.alpha

        # Find winner
        significant_results = [r for r in results if r["significant"] and r["lift"] > 0]
        if significant_results:
            winner = max(significant_results, key=lambda x: x["lift"])["name"]
        else:
            winner = "Control (no significant improvement)"

        return {
            "control": {
                "visitors": c_visitors,
                "conversions": c_conversions,
                "rate": c_rate
            },
            "variants": results,
            "winner": winner,
            "correction_method": correction,
            "alpha": self.alpha
        }

    def estimate_duration(self, daily_visitors: int, baseline_rate: float,
                         minimum_detectable_effect: float,
                         power: float = 0.8) -> Dict:
        """
        Estimate test duration based on traffic.

        Args:
            daily_visitors: Average daily visitors
            baseline_rate: Current conversion rate
            minimum_detectable_effect: Relative change to detect
            power: Desired statistical power

        Returns:
            Dictionary with duration estimates
        """
        sample = self.calculate_sample_size(
            baseline_rate, minimum_detectable_effect, power
        )

        # Assuming 50/50 split
        visitors_per_variant_per_day = daily_visitors / 2

        if visitors_per_variant_per_day > 0:
            days = math.ceil(sample["sample_size_per_variant"] / visitors_per_variant_per_day)
        else:
            days = float('inf')

        return {
            "days": days,
            "weeks": math.ceil(days / 7),
            "sample_size_per_variant": sample["sample_size_per_variant"],
            "daily_visitors": daily_visitors,
            "visitors_per_variant_per_day": visitors_per_variant_per_day
        }


def main():
    parser = argparse.ArgumentParser(
        description="A/B Test Calculator - Statistical significance testing"
    )

    parser.add_argument("--test", nargs=4, type=int, metavar=("CV", "CC", "VV", "VC"),
                       help="Test significance: control_visitors control_conversions variant_visitors variant_conversions")
    parser.add_argument("--sample-size", action="store_true",
                       help="Calculate sample size")
    parser.add_argument("--power-analysis", action="store_true",
                       help="Perform power analysis")
    parser.add_argument("--bayesian", nargs=4, type=int, metavar=("CV", "CC", "VV", "VC"),
                       help="Bayesian analysis: control_visitors control_conversions variant_visitors variant_conversions")
    parser.add_argument("--test-multi", nargs="+", type=int,
                       help="Test multiple variants: c_visitors c_conv v1_visitors v1_conv [v2_visitors v2_conv ...]")

    parser.add_argument("--baseline", type=float, help="Baseline conversion rate (e.g., 0.05)")
    parser.add_argument("--mde", type=float, help="Minimum detectable effect (e.g., 0.10)")
    parser.add_argument("--power", type=float, default=0.8, help="Statistical power (default: 0.8)")
    parser.add_argument("--samples", type=int, help="Sample size per variant")
    parser.add_argument("--alpha", type=float, default=0.05, help="Significance level (default: 0.05)")
    parser.add_argument("--method", choices=["chi_square", "z_test"], default="chi_square",
                       help="Test method (default: chi_square)")

    parser.add_argument("--json", action="store_true", help="Output as JSON")

    args = parser.parse_args()

    calc = ABTestCalculator(alpha=args.alpha)

    if args.test:
        result = calc.test_significance(
            args.test[0], args.test[1],
            args.test[2], args.test[3],
            test=args.method
        )

        if args.json:
            print(json.dumps(result, indent=2))
        else:
            print("\n=== A/B Test Results ===")
            print(f"Control:  {args.test[1]}/{args.test[0]} ({result['control_rate']:.2%})")
            print(f"Variant:  {args.test[3]}/{args.test[2]} ({result['variant_rate']:.2%})")
            print(f"\nLift: {result['lift']:+.2%} ({result['lift_absolute']:+.4f} absolute)")
            print(f"P-value: {result['p_value']:.4f}")
            print(f"Significant: {'Yes' if result['significant'] else 'No'} (alpha={args.alpha})")
            print(f"\n95% CI for lift: [{result['confidence_interval']['lower']:.2%}, {result['confidence_interval']['upper']:.2%}]")
            print(f"\n{result['recommendation']}")

    elif args.sample_size:
        if not args.baseline or not args.mde:
            parser.error("--sample-size requires --baseline and --mde")

        result = calc.calculate_sample_size(
            args.baseline, args.mde, args.power
        )

        if args.json:
            print(json.dumps(result, indent=2))
        else:
            print("\n=== Sample Size Calculation ===")
            print(f"Baseline rate: {result['baseline_rate']:.2%}")
            print(f"Expected variant rate: {result['expected_variant_rate']:.2%}")
            print(f"Minimum detectable effect: {result['minimum_detectable_effect']:.1%}")
            print(f"Power: {result['power']:.0%}")
            print(f"Alpha: {result['alpha']}")
            print(f"\nRequired sample size per variant: {result['sample_size_per_variant']:,}")
            print(f"Total sample size (both variants): {result['total_sample_size']:,}")

    elif args.power_analysis:
        if not args.baseline or not args.mde or not args.samples:
            parser.error("--power-analysis requires --baseline, --mde, and --samples")

        result = calc.calculate_power(
            args.baseline, args.mde, args.samples
        )

        if args.json:
            print(json.dumps(result, indent=2))
        else:
            print("\n=== Power Analysis ===")
            print(f"Sample size per variant: {result['sample_size_per_variant']:,}")
            print(f"Baseline rate: {result['baseline_rate']:.2%}")
            print(f"MDE: {result['minimum_detectable_effect']:.1%}")
            print(f"\nStatistical Power: {result['power']:.1%}")
            print(f"({result['interpretation']})")

    elif args.bayesian:
        result = calc.bayesian_analysis(
            args.bayesian[0], args.bayesian[1],
            args.bayesian[2], args.bayesian[3]
        )

        if args.json:
            print(json.dumps(result, indent=2))
        else:
            print("\n=== Bayesian Analysis ===")
            print(f"Control rate: {result['control_rate_mean']:.2%}")
            print(f"Variant rate: {result['variant_rate_mean']:.2%}")
            print(f"\nProbability variant beats control: {result['prob_variant_better']:.1%}")
            print(f"Expected lift: {result['expected_lift']:+.2%}")
            print(f"95% Credible interval: [{result['credible_interval_95'][0]:.2%}, {result['credible_interval_95'][1]:.2%}]")

    elif args.test_multi:
        if len(args.test_multi) < 4 or len(args.test_multi) % 2 != 0:
            parser.error("--test-multi requires pairs of (visitors, conversions)")

        control = (args.test_multi[0], args.test_multi[1])
        variants = []
        for i in range(2, len(args.test_multi), 2):
            variants.append((args.test_multi[i], args.test_multi[i+1]))

        result = calc.test_multiple_variants(control, variants)

        if args.json:
            print(json.dumps(result, indent=2))
        else:
            print("\n=== Multiple Variant Test ===")
            print(f"Control: {result['control']['conversions']}/{result['control']['visitors']} ({result['control']['rate']:.2%})")
            print(f"\nVariants:")
            for v in result['variants']:
                sig = "*" if v['significant'] else ""
                print(f"  {v['name']}: {v['conversions']}/{v['visitors']} ({v['rate']:.2%}) "
                      f"Lift: {v['lift']:+.2%} p={v['p_value']:.4f}{sig}")
            print(f"\nWinner: {result['winner']}")
            print(f"(Correction: {result['correction_method']})")

    else:
        parser.print_help()


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Budget Analyzer - Analyze expenses, track budgets, get savings recommendations.
"""

import argparse
import json
import re
from datetime import datetime
from typing import Dict, List, Optional, Tuple
from io import BytesIO

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import matplotlib

matplotlib.use('Agg')


class BudgetAnalyzer:
    """Analyze personal or business expenses."""

    DEFAULT_CATEGORIES = {
        "Food & Dining": ["restaurant", "cafe", "starbucks", "mcdonald", "uber eats",
                         "doordash", "grubhub", "chipotle", "subway", "pizza", "food"],
        "Transportation": ["uber", "lyft", "gas", "shell", "chevron", "parking",
                          "transit", "metro", "bus", "taxi", "fuel"],
        "Shopping": ["amazon", "walmart", "target", "costco", "best buy", "ebay",
                    "etsy", "shop", "store", "market"],
        "Utilities": ["electric", "water", "gas bill", "internet", "phone", "verizon",
                     "at&t", "comcast", "utility", "power"],
        "Entertainment": ["netflix", "spotify", "hulu", "disney", "movie", "theater",
                         "concert", "game", "steam"],
        "Healthcare": ["pharmacy", "cvs", "walgreens", "doctor", "hospital", "medical",
                      "dental", "vision", "health"],
        "Travel": ["airline", "hotel", "airbnb", "booking", "flight", "vacation",
                  "resort", "travel"],
        "Subscriptions": ["subscription", "membership", "monthly", "annual", "premium"],
        "Insurance": ["insurance", "geico", "state farm", "allstate", "progressive"],
        "Education": ["tuition", "school", "university", "course", "book", "udemy"],
    }

    def __init__(self):
        """Initialize the analyzer."""
        self.df = None
        self.categories = self.DEFAULT_CATEGORIES.copy()
        self.budget = {}
        self.date_col = "date"
        self.amount_col = "amount"
        self.description_col = "description"
        self.category_col = "category"

    def load_csv(self, filepath: str, date_col: str, amount_col: str,
                 description_col: str = None, category_col: str = None) -> 'BudgetAnalyzer':
        """Load transaction data from CSV."""
        self.df = pd.read_csv(filepath)
        self.date_col = date_col
        self.amount_col = amount_col
        self.description_col = description_col or "description"
        self.category_col = category_col or "category"

        # Parse dates
        self.df[self.date_col] = pd.to_datetime(self.df[self.date_col])

        # Ensure amount is numeric and positive (expenses)
        self.df[self.amount_col] = pd.to_numeric(self.df[self.amount_col], errors='coerce').abs()

        # Add category column if not present
        if self.category_col not in self.df.columns:
            self.df[self.category_col] = "Uncategorized"

        return self

    def load_dataframe(self, df: pd.DataFrame, date_col: str = "date",
                       amount_col: str = "amount", description_col: str = "description",
                       category_col: str = "category") -> 'BudgetAnalyzer':
        """Load from existing DataFrame."""
        self.df = df.copy()
        self.date_col = date_col
        self.amount_col = amount_col
        self.description_col = description_col
        self.category_col = category_col

        self.df[self.date_col] = pd.to_datetime(self.df[self.date_col])
        self.df[self.amount_col] = pd.to_numeric(self.df[self.amount_col], errors='coerce').abs()

        if self.category_col not in self.df.columns:
            self.df[self.category_col] = "Uncategorized"

        return self

    def set_categories(self, categories: Dict[str, List[str]]) -> 'BudgetAnalyzer':
        """Set custom category mappings."""
        self.categories = categories
        return self

    def auto_categorize(self) -> 'BudgetAnalyzer':
        """Auto-categorize transactions based on description."""
        if self.description_col not in self.df.columns:
            return self

        def categorize(description: str) -> str:
            if pd.isna(description):
                return "Uncategorized"
            desc_lower = str(description).lower()
            for category, keywords in self.categories.items():
                for keyword in keywords:
                    if keyword.lower() in desc_lower:
                        return category
            return "Other"

        self.df[self.category_col] = self.df[self.description_col].apply(categorize)
        return self

    def set_budget(self, budget: Dict[str, float]) -> 'BudgetAnalyzer':
        """Set budget targets by category."""
        self.budget = budget
        return self

    def analyze(self) -> Dict:
        """Get full analysis summary."""
        if self.df is None or len(self.df) == 0:
            return {}

        total = self.df[self.amount_col].sum()
        count = len(self.df)
        avg = self.df[self.amount_col].mean()
        largest_idx = self.df[self.amount_col].idxmax()
        largest = self.df.loc[largest_idx]

        categories = self.df.groupby(self.category_col)[self.amount_col].sum().to_dict()

        return {
            "total_spent": round(total, 2),
            "transaction_count": count,
            "date_range": {
                "start": self.df[self.date_col].min().strftime("%Y-%m-%d"),
                "end": self.df[self.date_col].max().strftime("%Y-%m-%d")
            },
            "average_transaction": round(avg, 2),
            "median_transaction": round(self.df[self.amount_col].median(), 2),
            "largest_expense": {
                "amount": round(largest[self.amount_col], 2),
                "description": largest.get(self.description_col, "N/A") if self.description_col in self.df.columns else "N/A",
                "date": largest[self.date_col].strftime("%Y-%m-%d")
            },
            "categories": {k: round(v, 2) for k, v in categories.items()}
        }

    def by_category(self) -> pd.DataFrame:
        """Get spending breakdown by category."""
        if self.df is None:
            return pd.DataFrame()

        grouped = self.df.groupby(self.category_col).agg({
            self.amount_col: ['sum', 'count', 'mean']
        }).reset_index()

        grouped.columns = ['category', 'amount', 'count', 'average']
        total = grouped['amount'].sum()
        grouped['percentage'] = (grouped['amount'] / total * 100).round(1)
        grouped = grouped.sort_values('amount', ascending=False)
        grouped['amount'] = grouped['amount'].round(2)
        grouped['average'] = grouped['average'].round(2)

        return grouped

    def by_month(self) -> pd.DataFrame:
        """Get spending by month."""
        if self.df is None:
            return pd.DataFrame()

        df = self.df.copy()
        df['month'] = df[self.date_col].dt.to_period('M')

        grouped = df.groupby('month').agg({
            self.amount_col: ['sum', 'count', 'mean']
        }).reset_index()

        grouped.columns = ['month', 'total', 'count', 'avg_transaction']
        grouped['month'] = grouped['month'].astype(str)
        grouped['total'] = grouped['total'].round(2)
        grouped['avg_transaction'] = grouped['avg_transaction'].round(2)

        return grouped

    def by_day_of_week(self) -> pd.DataFrame:
        """Get spending by day of week."""
        if self.df is None:
            return pd.DataFrame()

        df = self.df.copy()
        df['day_of_week'] = df[self.date_col].dt.day_name()

        grouped = df.groupby('day_of_week').agg({
            self.amount_col: ['sum', 'count', 'mean']
        }).reset_index()

        grouped.columns = ['day', 'total', 'count', 'average']

        # Sort by day of week
        days_order = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
        grouped['day'] = pd.Categorical(grouped['day'], categories=days_order, ordered=True)
        grouped = grouped.sort_values('day')
        grouped['total'] = grouped['total'].round(2)
        grouped['average'] = grouped['average'].round(2)

        return grouped

    def top_expenses(self, n: int = 10) -> pd.DataFrame:
        """Get top N expenses."""
        if self.df is None:
            return pd.DataFrame()

        cols = [self.date_col, self.amount_col]
        if self.description_col in self.df.columns:
            cols.append(self.description_col)
        if self.category_col in self.df.columns:
            cols.append(self.category_col)

        return self.df.nlargest(n, self.amount_col)[cols].reset_index(drop=True)

    def recurring_expenses(self, min_occurrences: int = 2) -> pd.DataFrame:
        """Identify recurring expenses."""
        if self.df is None or self.description_col not in self.df.columns:
            return pd.DataFrame()

        # Group by similar descriptions
        grouped = self.df.groupby(self.description_col).agg({
            self.amount_col: ['count', 'sum', 'mean', 'std']
        }).reset_index()

        grouped.columns = ['description', 'occurrences', 'total', 'average', 'std_dev']

        # Filter for recurring (multiple occurrences, similar amounts)
        recurring = grouped[grouped['occurrences'] >= min_occurrences].copy()
        recurring['likely_subscription'] = recurring['std_dev'].fillna(0) < (recurring['average'] * 0.1)
        recurring = recurring.sort_values('occurrences', ascending=False)

        return recurring

    def compare_periods(self, period1: str, period2: str) -> Dict:
        """Compare spending between two periods."""
        if self.df is None:
            return {}

        df = self.df.copy()
        df['period'] = df[self.date_col].dt.to_period('M').astype(str)

        # Support quarter format
        if 'Q' in period1:
            df['period'] = df[self.date_col].dt.to_period('Q').astype(str)

        p1_df = df[df['period'] == period1]
        p2_df = df[df['period'] == period2]

        p1_total = p1_df[self.amount_col].sum()
        p2_total = p2_df[self.amount_col].sum()
        diff = p2_total - p1_total
        pct_change = (diff / p1_total * 100) if p1_total > 0 else 0

        # Category comparison
        p1_cats = p1_df.groupby(self.category_col)[self.amount_col].sum()
        p2_cats = p2_df.groupby(self.category_col)[self.amount_col].sum()

        all_cats = set(p1_cats.index) | set(p2_cats.index)
        category_changes = {}
        for cat in all_cats:
            p1_val = p1_cats.get(cat, 0)
            p2_val = p2_cats.get(cat, 0)
            change = p2_val - p1_val
            pct = (change / p1_val * 100) if p1_val > 0 else (100 if p2_val > 0 else 0)
            category_changes[cat] = {
                "period1": round(p1_val, 2),
                "period2": round(p2_val, 2),
                "change": round(change, 2),
                "percent": round(pct, 1)
            }

        return {
            "period1": period1,
            "period2": period2,
            "period1_total": round(p1_total, 2),
            "period2_total": round(p2_total, 2),
            "difference": round(diff, 2),
            "percent_change": round(pct_change, 1),
            "category_changes": category_changes
        }

    def budget_vs_actual(self) -> pd.DataFrame:
        """Compare actual spending to budget."""
        if not self.budget or self.df is None:
            return pd.DataFrame()

        actual = self.df.groupby(self.category_col)[self.amount_col].sum()

        data = []
        for category, budget_amount in self.budget.items():
            actual_amount = actual.get(category, 0)
            diff = budget_amount - actual_amount
            status = "under" if diff >= 0 else "over"
            if diff >= 0 and actual_amount >= budget_amount * 0.9:
                status = "warning"

            data.append({
                "category": category,
                "budget": budget_amount,
                "actual": round(actual_amount, 2),
                "difference": round(diff, 2),
                "percent_used": round(actual_amount / budget_amount * 100, 1) if budget_amount > 0 else 0,
                "status": status
            })

        return pd.DataFrame(data).sort_values("actual", ascending=False)

    def budget_alerts(self) -> List[Dict]:
        """Get budget alerts for categories over or near budget."""
        comparison = self.budget_vs_actual()
        if comparison.empty:
            return []

        alerts = []
        for _, row in comparison.iterrows():
            if row['status'] == 'over':
                alerts.append({
                    "category": row['category'],
                    "status": "over",
                    "actual": row['actual'],
                    "budget": row['budget'],
                    "percent_over": round((row['actual'] - row['budget']) / row['budget'] * 100, 1)
                })
            elif row['status'] == 'warning':
                alerts.append({
                    "category": row['category'],
                    "status": "warning",
                    "actual": row['actual'],
                    "budget": row['budget'],
                    "percent_used": row['percent_used']
                })

        return alerts

    def get_recommendations(self) -> List[str]:
        """Get spending recommendations."""
        if self.df is None:
            return []

        recommendations = []

        # Category analysis
        categories = self.by_category()
        if not categories.empty:
            top_cat = categories.iloc[0]
            recommendations.append(
                f"Top spending category: {top_cat['category']} (${top_cat['amount']:.2f}, "
                f"{top_cat['percentage']:.1f}% of total)"
            )

        # Month-over-month changes
        monthly = self.by_month()
        if len(monthly) >= 2:
            last_two = monthly.tail(2)
            if len(last_two) == 2:
                change = last_two.iloc[1]['total'] - last_two.iloc[0]['total']
                pct = (change / last_two.iloc[0]['total'] * 100) if last_two.iloc[0]['total'] > 0 else 0
                if pct > 10:
                    recommendations.append(
                        f"Spending increased {pct:.1f}% from {last_two.iloc[0]['month']} to {last_two.iloc[1]['month']}. "
                        "Review recent purchases for unnecessary expenses."
                    )
                elif pct < -10:
                    recommendations.append(
                        f"Great job! Spending decreased {abs(pct):.1f}% from {last_two.iloc[0]['month']}."
                    )

        # Recurring expenses
        recurring = self.recurring_expenses()
        if not recurring.empty:
            subscriptions = recurring[recurring['likely_subscription'] == True]
            if not subscriptions.empty:
                total_subs = subscriptions['average'].sum()
                recommendations.append(
                    f"Found {len(subscriptions)} potential subscriptions totaling ~${total_subs:.2f}/period. "
                    "Review for unused services."
                )

        # Budget alerts
        alerts = self.budget_alerts()
        for alert in alerts:
            if alert['status'] == 'over':
                recommendations.append(
                    f"{alert['category']} is {alert['percent_over']:.1f}% over budget. "
                    f"Actual: ${alert['actual']:.2f}, Budget: ${alert['budget']:.2f}"
                )

        # Day of week patterns
        by_day = self.by_day_of_week()
        if not by_day.empty:
            highest_day = by_day.loc[by_day['total'].idxmax()]
            recommendations.append(
                f"Highest spending day: {highest_day['day']} (${highest_day['total']:.2f}). "
                "Consider planning purchases for other days."
            )

        return recommendations

    def spending_score(self) -> Dict:
        """Calculate overall spending health score."""
        if self.df is None:
            return {}

        scores = []

        # Budget adherence (if budget set)
        if self.budget:
            comparison = self.budget_vs_actual()
            if not comparison.empty:
                under_budget = len(comparison[comparison['status'] == 'under'])
                adherence = (under_budget / len(comparison)) * 100
                scores.append(('budget_adherence', adherence))

        # Spending consistency (low variance = more consistent)
        monthly = self.by_month()
        if len(monthly) >= 3:
            cv = monthly['total'].std() / monthly['total'].mean() if monthly['total'].mean() > 0 else 0
            consistency = max(0, 100 - cv * 100)
            scores.append(('spending_consistency', consistency))

        # Category diversity (not too concentrated)
        categories = self.by_category()
        if not categories.empty:
            top_pct = categories.iloc[0]['percentage']
            diversity = max(0, 100 - (top_pct - 30))  # Penalty if top category > 30%
            scores.append(('category_diversity', diversity))

        if not scores:
            return {"overall_score": 50, "grade": "N/A", "summary": "Insufficient data"}

        overall = np.mean([s[1] for s in scores])
        grade = 'A' if overall >= 90 else 'B' if overall >= 75 else 'C' if overall >= 60 else 'D' if overall >= 40 else 'F'

        return {
            "overall_score": round(overall),
            "factors": {name: round(score) for name, score in scores},
            "grade": grade,
            "summary": self._get_score_summary(grade)
        }

    def _get_score_summary(self, grade: str) -> str:
        """Get summary text for spending score."""
        summaries = {
            'A': "Excellent spending habits. Keep up the great work!",
            'B': "Good spending habits with minor room for improvement.",
            'C': "Average spending habits. Consider reviewing your budget.",
            'D': "Below average. Significant opportunities to improve spending.",
            'F': "Needs attention. Consider creating a strict budget plan."
        }
        return summaries.get(grade, "Unable to assess spending habits.")

    def plot_categories(self, output: str) -> str:
        """Create pie chart of spending by category."""
        categories = self.by_category()
        if categories.empty:
            return ""

        fig, ax = plt.subplots(figsize=(10, 8))

        # Limit to top 8 categories, group rest as "Other"
        if len(categories) > 8:
            top = categories.head(7)
            other_total = categories.iloc[7:]['amount'].sum()
            other_row = pd.DataFrame([{
                'category': 'Other',
                'amount': other_total,
                'percentage': categories.iloc[7:]['percentage'].sum()
            }])
            categories = pd.concat([top, other_row], ignore_index=True)

        colors = plt.cm.Set3(np.linspace(0, 1, len(categories)))
        wedges, texts, autotexts = ax.pie(
            categories['amount'],
            labels=categories['category'],
            autopct='%1.1f%%',
            colors=colors,
            startangle=90
        )

        ax.set_title('Spending by Category', fontsize=14, fontweight='bold')
        plt.tight_layout()
        plt.savefig(output, dpi=150, bbox_inches='tight')
        plt.close()

        return output

    def plot_trends(self, output: str) -> str:
        """Create line chart of monthly spending."""
        monthly = self.by_month()
        if monthly.empty:
            return ""

        fig, ax = plt.subplots(figsize=(12, 6))

        ax.plot(monthly['month'], monthly['total'], marker='o', linewidth=2, markersize=8)
        ax.fill_between(monthly['month'], monthly['total'], alpha=0.3)

        ax.set_xlabel('Month', fontsize=12)
        ax.set_ylabel('Total Spending ($)', fontsize=12)
        ax.set_title('Monthly Spending Trend', fontsize=14, fontweight='bold')

        plt.xticks(rotation=45, ha='right')
        ax.grid(True, alpha=0.3)

        # Add average line
        avg = monthly['total'].mean()
        ax.axhline(y=avg, color='red', linestyle='--', alpha=0.7, label=f'Average: ${avg:.2f}')
        ax.legend()

        plt.tight_layout()
        plt.savefig(output, dpi=150, bbox_inches='tight')
        plt.close()

        return output

    def plot_budget_comparison(self, output: str) -> str:
        """Create bar chart comparing budget vs actual."""
        comparison = self.budget_vs_actual()
        if comparison.empty:
            return ""

        fig, ax = plt.subplots(figsize=(12, 6))

        x = np.arange(len(comparison))
        width = 0.35

        bars1 = ax.bar(x - width/2, comparison['budget'], width, label='Budget', color='steelblue')
        bars2 = ax.bar(x + width/2, comparison['actual'], width, label='Actual', color='coral')

        ax.set_xlabel('Category', fontsize=12)
        ax.set_ylabel('Amount ($)', fontsize=12)
        ax.set_title('Budget vs Actual Spending', fontsize=14, fontweight='bold')
        ax.set_xticks(x)
        ax.set_xticklabels(comparison['category'], rotation=45, ha='right')
        ax.legend()

        ax.grid(True, alpha=0.3, axis='y')

        plt.tight_layout()
        plt.savefig(output, dpi=150, bbox_inches='tight')
        plt.close()

        return output

    def generate_report(self, output: str, format: str = "pdf") -> str:
        """Generate comprehensive report."""
        if format == "html":
            return self._generate_html_report(output)
        else:
            return self._generate_pdf_report(output)

    def _generate_pdf_report(self, output: str) -> str:
        """Generate PDF report using reportlab."""
        from reportlab.lib import colors
        from reportlab.lib.pagesizes import letter
        from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
        from reportlab.lib.units import inch
        from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle, Image

        doc = SimpleDocTemplate(output, pagesize=letter)
        styles = getSampleStyleSheet()
        story = []

        # Title
        title_style = ParagraphStyle(
            'Title',
            parent=styles['Heading1'],
            fontSize=24,
            spaceAfter=30
        )
        story.append(Paragraph("Budget Analysis Report", title_style))
        story.append(Spacer(1, 20))

        # Summary
        summary = self.analyze()
        if summary:
            story.append(Paragraph("Executive Summary", styles['Heading2']))
            summary_text = f"""
            <b>Total Spent:</b> ${summary['total_spent']:,.2f}<br/>
            <b>Transactions:</b> {summary['transaction_count']}<br/>
            <b>Average Transaction:</b> ${summary['average_transaction']:,.2f}<br/>
            <b>Date Range:</b> {summary['date_range']['start']} to {summary['date_range']['end']}<br/>
            <b>Largest Expense:</b> ${summary['largest_expense']['amount']:,.2f}
            """
            story.append(Paragraph(summary_text, styles['Normal']))
            story.append(Spacer(1, 20))

        # Category breakdown
        categories = self.by_category()
        if not categories.empty:
            story.append(Paragraph("Spending by Category", styles['Heading2']))

            table_data = [['Category', 'Amount', 'Percentage', 'Count']]
            for _, row in categories.head(10).iterrows():
                table_data.append([
                    row['category'],
                    f"${row['amount']:,.2f}",
                    f"{row['percentage']:.1f}%",
                    str(int(row['count']))
                ])

            table = Table(table_data, colWidths=[2*inch, 1.5*inch, 1*inch, 1*inch])
            table.setStyle(TableStyle([
                ('BACKGROUND', (0, 0), (-1, 0), colors.steelblue),
                ('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke),
                ('ALIGN', (0, 0), (-1, -1), 'CENTER'),
                ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'),
                ('FONTSIZE', (0, 0), (-1, 0), 12),
                ('BOTTOMPADDING', (0, 0), (-1, 0), 12),
                ('BACKGROUND', (0, 1), (-1, -1), colors.beige),
                ('GRID', (0, 0), (-1, -1), 1, colors.black)
            ]))
            story.append(table)
            story.append(Spacer(1, 20))

        # Budget comparison
        budget_comp = self.budget_vs_actual()
        if not budget_comp.empty:
            story.append(Paragraph("Budget vs Actual", styles['Heading2']))

            table_data = [['Category', 'Budget', 'Actual', 'Difference', 'Status']]
            for _, row in budget_comp.iterrows():
                status_color = 'green' if row['status'] == 'under' else 'orange' if row['status'] == 'warning' else 'red'
                table_data.append([
                    row['category'],
                    f"${row['budget']:,.2f}",
                    f"${row['actual']:,.2f}",
                    f"${row['difference']:,.2f}",
                    row['status'].upper()
                ])

            table = Table(table_data, colWidths=[1.5*inch, 1.2*inch, 1.2*inch, 1.2*inch, 1*inch])
            table.setStyle(TableStyle([
                ('BACKGROUND', (0, 0), (-1, 0), colors.steelblue),
                ('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke),
                ('ALIGN', (0, 0), (-1, -1), 'CENTER'),
                ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'),
                ('GRID', (0, 0), (-1, -1), 1, colors.black)
            ]))
            story.append(table)
            story.append(Spacer(1, 20))

        # Recommendations
        recommendations = self.get_recommendations()
        if recommendations:
            story.append(Paragraph("Recommendations", styles['Heading2']))
            for rec in recommendations:
                story.append(Paragraph(f"• {rec}", styles['Normal']))
            story.append(Spacer(1, 10))

        # Spending score
        score = self.spending_score()
        if score:
            story.append(Paragraph("Spending Health Score", styles['Heading2']))
            score_text = f"""
            <b>Overall Score:</b> {score['overall_score']}/100 (Grade: {score['grade']})<br/>
            <b>Summary:</b> {score['summary']}
            """
            story.append(Paragraph(score_text, styles['Normal']))

        doc.build(story)
        return output

    def _generate_html_report(self, output: str) -> str:
        """Generate HTML report."""
        summary = self.analyze()
        categories = self.by_category()
        monthly = self.by_month()
        recommendations = self.get_recommendations()
        score = self.spending_score()

        html = f"""<!DOCTYPE html>
<html>
<head>
    <title>Budget Analysis Report</title>
    <style>
        body {{ font-family: Arial, sans-serif; max-width: 900px; margin: 0 auto; padding: 20px; }}
        h1 {{ color: #2c3e50; }}
        h2 {{ color: #34495e; border-bottom: 2px solid #3498db; padding-bottom: 10px; }}
        table {{ width: 100%; border-collapse: collapse; margin: 20px 0; }}
        th {{ background-color: #3498db; color: white; padding: 12px; text-align: left; }}
        td {{ padding: 10px; border-bottom: 1px solid #ddd; }}
        tr:hover {{ background-color: #f5f5f5; }}
        .summary {{ background: #ecf0f1; padding: 20px; border-radius: 8px; }}
        .score {{ font-size: 48px; font-weight: bold; color: #27ae60; }}
        .recommendation {{ background: #fff3cd; padding: 10px; margin: 5px 0; border-left: 4px solid #ffc107; }}
        .over {{ color: #e74c3c; }}
        .under {{ color: #27ae60; }}
        .warning {{ color: #f39c12; }}
    </style>
</head>
<body>
    <h1>Budget Analysis Report</h1>

    <div class="summary">
        <h2>Executive Summary</h2>
        <p><strong>Total Spent:</strong> ${summary.get('total_spent', 0):,.2f}</p>
        <p><strong>Transactions:</strong> {summary.get('transaction_count', 0)}</p>
        <p><strong>Average Transaction:</strong> ${summary.get('average_transaction', 0):,.2f}</p>
        <p><strong>Date Range:</strong> {summary.get('date_range', {}).get('start', 'N/A')} to {summary.get('date_range', {}).get('end', 'N/A')}</p>
    </div>

    <h2>Spending by Category</h2>
    <table>
        <tr><th>Category</th><th>Amount</th><th>Percentage</th><th>Transactions</th></tr>
"""

        for _, row in categories.iterrows():
            html += f"<tr><td>{row['category']}</td><td>${row['amount']:,.2f}</td><td>{row['percentage']:.1f}%</td><td>{int(row['count'])}</td></tr>\n"

        html += """
    </table>

    <h2>Monthly Trends</h2>
    <table>
        <tr><th>Month</th><th>Total</th><th>Transactions</th><th>Average</th></tr>
"""

        for _, row in monthly.iterrows():
            html += f"<tr><td>{row['month']}</td><td>${row['total']:,.2f}</td><td>{int(row['count'])}</td><td>${row['avg_transaction']:,.2f}</td></tr>\n"

        html += """
    </table>

    <h2>Recommendations</h2>
"""
        for rec in recommendations:
            html += f'<div class="recommendation">{rec}</div>\n'

        html += f"""
    <h2>Spending Health Score</h2>
    <div class="score">{score.get('overall_score', 'N/A')}/100</div>
    <p><strong>Grade:</strong> {score.get('grade', 'N/A')}</p>
    <p>{score.get('summary', '')}</p>

</body>
</html>
"""

        with open(output, 'w') as f:
            f.write(html)

        return output

    def to_csv(self, output: str) -> str:
        """Export analyzed data to CSV."""
        if self.df is not None:
            self.df.to_csv(output, index=False)
        return output


def main():
    parser = argparse.ArgumentParser(description="Budget Analyzer - Analyze expenses and track budgets")

    parser.add_argument("--input", "-i", required=True, help="Input CSV file")
    parser.add_argument("--date", "-d", default="date", help="Date column name")
    parser.add_argument("--amount", "-a", default="amount", help="Amount column name")
    parser.add_argument("--description", default="description", help="Description column name")
    parser.add_argument("--category-col", help="Category column name (if pre-categorized)")

    parser.add_argument("--categories", help="JSON file with custom categories")
    parser.add_argument("--budget", help="JSON file with budget targets")

    parser.add_argument("--compare", nargs=2, metavar=('PERIOD1', 'PERIOD2'),
                       help="Compare two periods (e.g., 2024-01 2024-02)")

    parser.add_argument("--report", "-r", help="Generate report (PDF or HTML)")
    parser.add_argument("--plot-categories", help="Save category pie chart")
    parser.add_argument("--plot-trends", help="Save trends line chart")
    parser.add_argument("--plot-budget", help="Save budget comparison chart")

    parser.add_argument("--output", "-o", help="Output CSV with categories")

    args = parser.parse_args()

    analyzer = BudgetAnalyzer()

    # Load data
    analyzer.load_csv(
        args.input,
        date_col=args.date,
        amount_col=args.amount,
        description_col=args.description,
        category_col=args.category_col
    )

    # Load custom categories
    if args.categories:
        with open(args.categories) as f:
            analyzer.set_categories(json.load(f))

    # Auto-categorize if no category column
    if not args.category_col:
        analyzer.auto_categorize()

    # Load budget
    if args.budget:
        with open(args.budget) as f:
            analyzer.set_budget(json.load(f))

    # Period comparison
    if args.compare:
        comparison = analyzer.compare_periods(args.compare[0], args.compare[1])
        print(f"\n=== Period Comparison: {args.compare[0]} vs {args.compare[1]} ===")
        print(f"Period 1 Total: ${comparison['period1_total']:,.2f}")
        print(f"Period 2 Total: ${comparison['period2_total']:,.2f}")
        print(f"Difference: ${comparison['difference']:,.2f} ({comparison['percent_change']:+.1f}%)")
    else:
        # Show summary
        summary = analyzer.analyze()
        print("\n=== Budget Analysis Summary ===")
        print(f"Total Spent: ${summary['total_spent']:,.2f}")
        print(f"Transactions: {summary['transaction_count']}")
        print(f"Average: ${summary['average_transaction']:,.2f}")
        print(f"Date Range: {summary['date_range']['start']} to {summary['date_range']['end']}")

        print("\n=== Top Categories ===")
        categories = analyzer.by_category()
        for _, row in categories.head(5).iterrows():
            print(f"  {row['category']}: ${row['amount']:,.2f} ({row['percentage']:.1f}%)")

        if analyzer.budget:
            print("\n=== Budget Status ===")
            budget_comp = analyzer.budget_vs_actual()
            for _, row in budget_comp.iterrows():
                status_icon = "✓" if row['status'] == 'under' else "⚠" if row['status'] == 'warning' else "✗"
                print(f"  {status_icon} {row['category']}: ${row['actual']:,.2f} / ${row['budget']:,.2f}")

        print("\n=== Recommendations ===")
        for rec in analyzer.get_recommendations():
            print(f"  • {rec}")

        score = analyzer.spending_score()
        print(f"\n=== Spending Score: {score['overall_score']}/100 (Grade: {score['grade']}) ===")
        print(f"  {score['summary']}")

    # Generate visualizations
    if args.plot_categories:
        analyzer.plot_categories(args.plot_categories)
        print(f"\nSaved category chart: {args.plot_categories}")

    if args.plot_trends:
        analyzer.plot_trends(args.plot_trends)
        print(f"Saved trends chart: {args.plot_trends}")

    if args.plot_budget and analyzer.budget:
        analyzer.plot_budget_comparison(args.plot_budget)
        print(f"Saved budget chart: {args.plot_budget}")

    # Generate report
    if args.report:
        fmt = "html" if args.report.endswith(".html") else "pdf"
        analyzer.generate_report(args.report, format=fmt)
        print(f"\nGenerated report: {args.report}")

    # Export
    if args.output:
        analyzer.to_csv(args.output)
        print(f"Exported to: {args.output}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Correlation Explorer - Find and visualize correlations in datasets.

Features:
- Correlation matrix computation
- Multiple methods (Pearson, Spearman, Kendall)
- Heatmap visualization
- P-value significance testing
- Strong/weak correlation discovery
"""

import argparse
import json
from pathlib import Path
from typing import Dict, List, Optional, Tuple, Union


class CorrelationExplorer:
    """Analyze correlations between variables in datasets."""

    def __init__(self):
        """Initialize explorer."""
        self._pd = None
        self._np = None
        self._load_dependencies()
        self.data = None
        self._corr_matrix = None

    def _load_dependencies(self):
        """Load required libraries."""
        try:
            import pandas as pd
            import numpy as np
            self._pd = pd
            self._np = np
        except ImportError:
            raise ImportError("pandas and numpy required. Install with: pip install pandas numpy")

    def load_csv(self, filepath: str, **kwargs) -> 'CorrelationExplorer':
        """
        Load data from CSV file.

        Args:
            filepath: Path to CSV file
            **kwargs: Additional pandas read_csv args

        Returns:
            self for chaining
        """
        self.data = self._pd.read_csv(filepath, **kwargs)
        # Select only numeric columns
        self.data = self.data.select_dtypes(include=[self._np.number])
        self._corr_matrix = None
        return self

    def load_dataframe(self, df) -> 'CorrelationExplorer':
        """
        Load data from DataFrame.

        Args:
            df: pandas DataFrame

        Returns:
            self for chaining
        """
        self.data = df.select_dtypes(include=[self._np.number])
        self._corr_matrix = None
        return self

    def correlation_matrix(self, method: str = "pearson"):
        """
        Compute correlation matrix.

        Args:
            method: Correlation method (pearson, spearman, kendall)

        Returns:
            Correlation DataFrame
        """
        if self.data is None:
            raise ValueError("No data loaded. Use load_csv() or load_dataframe() first.")

        self._corr_matrix = self.data.corr(method=method)
        return self._corr_matrix

    def correlation_with_pvalues(self, method: str = "pearson") -> Tuple:
        """
        Compute correlation matrix with p-values.

        Args:
            method: Correlation method

        Returns:
            (correlation_matrix, pvalue_matrix)
        """
        from scipy import stats

        if self.data is None:
            raise ValueError("No data loaded.")

        n = len(self.data.columns)
        corr_matrix = self._pd.DataFrame(
            self._np.zeros((n, n)),
            columns=self.data.columns,
            index=self.data.columns
        )
        pval_matrix = corr_matrix.copy()

        for i, col1 in enumerate(self.data.columns):
            for j, col2 in enumerate(self.data.columns):
                # Remove NaN pairs
                mask = ~(self.data[col1].isna() | self.data[col2].isna())
                x = self.data[col1][mask]
                y = self.data[col2][mask]

                if method == 'pearson':
                    corr, pval = stats.pearsonr(x, y)
                elif method == 'spearman':
                    corr, pval = stats.spearmanr(x, y)
                elif method == 'kendall':
                    corr, pval = stats.kendalltau(x, y)
                else:
                    raise ValueError(f"Unknown method: {method}")

                corr_matrix.iloc[i, j] = corr
                pval_matrix.iloc[i, j] = pval

        self._corr_matrix = corr_matrix
        return corr_matrix, pval_matrix

    def correlate_with_target(self, target: str, method: str = "pearson"):
        """
        Get correlations of all variables with a target variable.

        Args:
            target: Target column name
            method: Correlation method

        Returns:
            Series of correlations sorted by absolute value
        """
        if self.data is None:
            raise ValueError("No data loaded.")

        if target not in self.data.columns:
            raise ValueError(f"Target '{target}' not found in data")

        corr_matrix = self.correlation_matrix(method)
        target_corr = corr_matrix[target].drop(target)
        return target_corr.sort_values(key=abs, ascending=False)

    def find_strong_correlations(
        self,
        threshold: float = 0.7,
        method: str = "pearson"
    ) -> List[Dict]:
        """
        Find pairs with strong correlations.

        Args:
            threshold: Minimum absolute correlation
            method: Correlation method

        Returns:
            List of correlation pairs above threshold
        """
        corr_matrix = self.correlation_matrix(method) if self._corr_matrix is None else self._corr_matrix
        strong = []

        for i in range(len(corr_matrix.columns)):
            for j in range(i + 1, len(corr_matrix.columns)):
                corr = corr_matrix.iloc[i, j]
                if abs(corr) >= threshold:
                    strong.append({
                        'var1': corr_matrix.columns[i],
                        'var2': corr_matrix.columns[j],
                        'correlation': round(corr, 4),
                        'abs_corr': round(abs(corr), 4)
                    })

        return sorted(strong, key=lambda x: x['abs_corr'], reverse=True)

    def find_weak_correlations(
        self,
        threshold: float = 0.3,
        method: str = "pearson"
    ) -> List[Dict]:
        """
        Find pairs with weak correlations.

        Args:
            threshold: Maximum absolute correlation
            method: Correlation method

        Returns:
            List of correlation pairs below threshold
        """
        corr_matrix = self.correlation_matrix(method) if self._corr_matrix is None else self._corr_matrix
        weak = []

        for i in range(len(corr_matrix.columns)):
            for j in range(i + 1, len(corr_matrix.columns)):
                corr = corr_matrix.iloc[i, j]
                if abs(corr) <= threshold:
                    weak.append({
                        'var1': corr_matrix.columns[i],
                        'var2': corr_matrix.columns[j],
                        'correlation': round(corr, 4),
                        'abs_corr': round(abs(corr), 4)
                    })

        return sorted(weak, key=lambda x: x['abs_corr'])

    def plot_heatmap(
        self,
        output: str,
        method: str = "pearson",
        cmap: str = "coolwarm",
        annot: bool = True,
        figsize: Tuple[int, int] = (10, 8),
        title: str = "Correlation Matrix",
        vmin: float = -1,
        vmax: float = 1
    ) -> str:
        """
        Generate correlation heatmap.

        Args:
            output: Output file path
            method: Correlation method
            cmap: Color map
            annot: Show correlation values
            figsize: Figure size
            title: Plot title
            vmin, vmax: Color scale limits

        Returns:
            Path to saved image
        """
        try:
            import matplotlib.pyplot as plt
            import seaborn as sns
        except ImportError:
            raise ImportError("matplotlib and seaborn required for plotting")

        corr_matrix = self.correlation_matrix(method) if self._corr_matrix is None else self._corr_matrix

        fig, ax = plt.subplots(figsize=figsize)
        sns.heatmap(
            corr_matrix,
            annot=annot,
            cmap=cmap,
            vmin=vmin,
            vmax=vmax,
            center=0,
            square=True,
            linewidths=0.5,
            fmt='.2f',
            ax=ax
        )
        ax.set_title(title)
        plt.tight_layout()
        plt.savefig(output, dpi=150)
        plt.close()

        return output

    def plot_scatter(
        self,
        var1: str,
        var2: str,
        output: str,
        figsize: Tuple[int, int] = (8, 6)
    ) -> str:
        """
        Create scatter plot of two variables.

        Args:
            var1: First variable
            var2: Second variable
            output: Output file path
            figsize: Figure size

        Returns:
            Path to saved image
        """
        try:
            import matplotlib.pyplot as plt
            import seaborn as sns
        except ImportError:
            raise ImportError("matplotlib and seaborn required for plotting")

        if self.data is None:
            raise ValueError("No data loaded.")

        fig, ax = plt.subplots(figsize=figsize)
        sns.scatterplot(data=self.data, x=var1, y=var2, ax=ax)

        # Add trend line
        from scipy import stats
        mask = ~(self.data[var1].isna() | self.data[var2].isna())
        x = self.data[var1][mask]
        y = self.data[var2][mask]
        slope, intercept, r, p, se = stats.linregress(x, y)
        ax.plot(x, slope * x + intercept, 'r--', label=f'r={r:.3f}')
        ax.legend()

        ax.set_title(f'{var1} vs {var2}')
        plt.tight_layout()
        plt.savefig(output, dpi=150)
        plt.close()

        return output

    def to_csv(self, output: str, method: str = "pearson") -> str:
        """Save correlation matrix to CSV."""
        corr_matrix = self.correlation_matrix(method) if self._corr_matrix is None else self._corr_matrix
        corr_matrix.to_csv(output)
        return output

    def to_json(self, output: str, method: str = "pearson") -> str:
        """Save correlation matrix to JSON."""
        corr_matrix = self.correlation_matrix(method) if self._corr_matrix is None else self._corr_matrix
        corr_matrix.to_json(output, orient='index', indent=2)
        return output


def main():
    """CLI entry point."""
    parser = argparse.ArgumentParser(description='Explore correlations in data')
    parser.add_argument('--input', '-i', required=True, help='Input CSV file')
    parser.add_argument('--output', '-o', help='Output file for correlation matrix')
    parser.add_argument('--heatmap', help='Generate heatmap image')
    parser.add_argument('--method', '-m', default='pearson',
                        choices=['pearson', 'spearman', 'kendall'])
    parser.add_argument('--target', '-t', help='Show correlations with target variable')
    parser.add_argument('--strong', action='store_true', help='Find strong correlations')
    parser.add_argument('--weak', action='store_true', help='Find weak correlations')
    parser.add_argument('--threshold', type=float, default=0.7,
                        help='Threshold for strong/weak (default: 0.7)')
    parser.add_argument('--pvalues', action='store_true', help='Include p-values')
    parser.add_argument('--json', action='store_true', help='Output as JSON')

    args = parser.parse_args()
    explorer = CorrelationExplorer()
    explorer.load_csv(args.input)

    if args.heatmap:
        explorer.plot_heatmap(args.heatmap, method=args.method)
        print(f"Heatmap saved to: {args.heatmap}")

    elif args.target:
        corr = explorer.correlate_with_target(args.target, method=args.method)
        print(f"\nCorrelations with '{args.target}':")
        print("-" * 40)
        for var, value in corr.items():
            print(f"  {var}: {value:.4f}")

    elif args.strong:
        strong = explorer.find_strong_correlations(args.threshold, method=args.method)
        print(f"\nStrong correlations (|r| >= {args.threshold}):")
        print("-" * 50)
        if strong:
            for pair in strong:
                print(f"  {pair['var1']} <-> {pair['var2']}: {pair['correlation']:.4f}")
        else:
            print("  No strong correlations found")

    elif args.weak:
        weak = explorer.find_weak_correlations(args.threshold, method=args.method)
        print(f"\nWeak correlations (|r| <= {args.threshold}):")
        print("-" * 50)
        for pair in weak[:20]:  # Show first 20
            print(f"  {pair['var1']} <-> {pair['var2']}: {pair['correlation']:.4f}")

    elif args.pvalues:
        corr, pvals = explorer.correlation_with_pvalues(method=args.method)
        print("\nCorrelation Matrix:")
        print(corr.round(4))
        print("\nP-Values:")
        print(pvals.round(4))

    else:
        corr_matrix = explorer.correlation_matrix(method=args.method)

        if args.output:
            if args.json:
                explorer.to_json(args.output)
            else:
                explorer.to_csv(args.output)
            print(f"Correlation matrix saved to: {args.output}")
        else:
            print("\nCorrelation Matrix:")
            print(corr_matrix.round(4))


if __name__ == "__main__":
    main()