import pandas as pd
import os
import glob

batch_folder = "/opt/airflow/batches_for_5years"
transformed_folder = "/opt/airflow/transformed_batches"
os.makedirs(transformed_folder, exist_ok=True)

batch_files = glob.glob(os.path.join(batch_folder, "us_rates_*.csv"))

for batch_file in batch_files:
    df = pd.read_csv(batch_file, parse_dates=['Date'])

    df = df.dropna(subset=['Date', 'Base', 'Currency', 'Rate'])
    df = df.drop_duplicates(subset=['Date', 'Base', 'Currency'])

    df['year'] = df['Date'].dt.year
    df['month'] = df['Date'].dt.month
    df['day_of_week'] = df['Date'].dt.dayofweek
    df['is_weekend'] = df['day_of_week'].isin([5, 6]).astype(int)

    df['currency_pair'] = df['Base'] + '/' + df['Currency']
    major_currencies = ['USD', 'EUR', 'JPY',
                        'GBP', 'CHF', 'AUD', 'CAD', 'NZD', 'CNY']
    df['is_major'] = df['Currency'].isin(major_currencies).astype(int)

    df = df.sort_values(['currency_pair', 'Date'])
    df['Rate_diff_1d'] = df.groupby('currency_pair')['Rate'].diff()
    df['Rate_pct_change_1d'] = df.groupby('currency_pair')['Rate'].pct_change()
    df['Rate_rolling_7d_mean'] = df.groupby('currency_pair')['Rate'].rolling(
        7, min_periods=1).mean().reset_index(0, drop=True)

    df = df[['Date', 'Base', 'Currency', 'Rate', 'currency_pair',
             'year', 'month', 'day_of_week', 'is_weekend',
             'is_major', 'Rate_diff_1d', 'Rate_pct_change_1d', 'Rate_rolling_7d_mean']]

    batch_name = os.path.basename(batch_file)
    transformed_file = os.path.join(
        transformed_folder, f"transformed_{batch_name}")
    df.to_csv(transformed_file, index=False)
    print(f"Transformed batch saved: {transformed_file}")