TariffTech/utils.py at main · pumpkins25/TariffTech · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
import pandas as pd
import numpy as np

def calculate_tariff_impact(tariff_change, category, tariff_data, price_data):
    """
    Calculate the estimated price impact of a tariff change based on historical data.

    Parameters:
    - tariff_change: Change in tariff rate (percentage points)
    - category: Product category to analyze
    - tariff_data: Historical tariff data
    - price_data: Historical price data

    Returns:
    - Estimated percentage price change
    """
    # Filter data for the specified category
    cat_tariff_data = tariff_data[tariff_data['Category'] == category]
    cat_price_data = price_data[price_data['Category'] == category]

    if cat_tariff_data.empty or cat_price_data.empty:
        # No data available for this category
        return 0.0

    # Calculate historical sensitivity
    try:
        # Group by year to get annual average tariff and price
        tariff_by_year = cat_tariff_data.groupby('Year')['Tariff_Rate'].mean().reset_index()
        price_by_year = cat_price_data.groupby('Year')['Price_Index'].mean().reset_index()

        # Merge the data
        historical_data = pd.merge(
            tariff_by_year,
            price_by_year,
            on='Year',
            how='inner'
        )

        if len(historical_data) <= 1:
            # Not enough data points for regression
            # Use a default sensitivity based on average across categories
            return tariff_change * 0.75

        # Calculate year-over-year changes
        historical_data['Tariff_Change'] = historical_data['Tariff_Rate'].diff()
        historical_data['Price_Change_Pct'] = historical_data['Price_Index'].pct_change() * 100

        # Remove the first row with NaN values and any rows with zero tariff change
        historical_data = historical_data.dropna()
        historical_data = historical_data[abs(historical_data['Tariff_Change']) > 0.001]

        if len(historical_data) == 0:
            # No valid data points after filtering
            return tariff_change * 0.75

        # Calculate sensitivity factors (price change % per 1% point tariff change)
        historical_data['Sensitivity'] = historical_data['Price_Change_Pct'] / historical_data['Tariff_Change']

        # Use median sensitivity to reduce impact of outliers
        median_sensitivity = historical_data['Sensitivity'].median()

        # Apply sensitivity to calculate expected price impact
        expected_impact = tariff_change * median_sensitivity

        return expected_impact

    except Exception as e:
        # Fallback to a reasonable estimate if calculation fails
        print(f"Error calculating impact for {category}: {str(e)}")
        return tariff_change * 0.75  # Assume 75% pass-through as a fallback

def get_year_range(data):
    """
    Get the minimum and maximum years from a dataset.
    """
    if 'Year' in data.columns:
        return data['Year'].min(), data['Year'].max()
    else:
        return None, None

def format_percentage(value):
    """
    Format a number as a percentage string.
    """
    return f"{value:.2f}%"

def calculate_correlation(tariff_series, price_series):
    """
    Calculate correlation between tariff rates and prices.
    """
    if len(tariff_series) != len(price_series):
        return None

    if len(tariff_series) <= 1:
        return None

    return tariff_series.corr(price_series)