src/core/fred_client.py

#!/usr/bin/env python3
"""
FRED Data Collector v2
A tool for collecting and analyzing Federal Reserve Economic Data (FRED)
using direct API calls instead of the fredapi library
"""

import os
import warnings
from datetime import datetime, timedelta

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import requests
import seaborn as sns

warnings.filterwarnings("ignore")

import os
import sys

sys.path.append(os.path.join(os.path.dirname(__file__), "..", ".."))

from config.settings import (DEFAULT_END_DATE, DEFAULT_START_DATE,
                             FRED_API_KEY, OUTPUT_DIR, PLOTS_DIR)


class FREDDataCollectorV2:
    def __init__(self, api_key=None):
        """Initialize the FRED data collector with API key."""
        self.api_key = api_key or FRED_API_KEY
        self.base_url = "https://api.stlouisfed.org/fred"

        # Create output directories
        os.makedirs(OUTPUT_DIR, exist_ok=True)
        os.makedirs(PLOTS_DIR, exist_ok=True)

        # Common economic indicators
        self.indicators = {
            "GDP": "GDP",  # Gross Domestic Product
            "UNRATE": "UNRATE",  # Unemployment Rate
            "CPIAUCSL": "CPIAUCSL",  # Consumer Price Index
            "FEDFUNDS": "FEDFUNDS",  # Federal Funds Rate
            "DGS10": "DGS10",  # 10-Year Treasury Rate
            "DEXUSEU": "DEXUSEU",  # US/Euro Exchange Rate
            "PAYEMS": "PAYEMS",  # Total Nonfarm Payrolls
            "INDPRO": "INDPRO",  # Industrial Production
            "M2SL": "M2SL",  # M2 Money Stock
            "PCE": "PCE",  # Personal Consumption Expenditures
        }

    def get_series_info(self, series_id):
        """Get information about a FRED series."""
        try:
            url = f"{self.base_url}/series"
            params = {
                "series_id": series_id,
                "api_key": self.api_key,
                "file_type": "json",
            }

            response = requests.get(url, params=params)

            if response.status_code == 200:
                data = response.json()
                series = data.get("seriess", [])

                if series:
                    s = series[0]
                    return {
                        "id": s["id"],
                        "title": s["title"],
                        "units": s.get("units", ""),
                        "frequency": s.get("frequency", ""),
                        "last_updated": s.get("last_updated", ""),
                        "notes": s.get("notes", ""),
                    }

            return None

        except Exception as e:
            print(f"Error getting info for {series_id}: {e}")
            return None

    def get_economic_data(self, series_ids, start_date=None, end_date=None):
        """Fetch economic data for specified series."""
        start_date = start_date or DEFAULT_START_DATE
        end_date = end_date or DEFAULT_END_DATE

        data = {}

        for series_id in series_ids:
            try:
                print(f"Fetching data for {series_id}...")

                url = f"{self.base_url}/series/observations"
                params = {
                    "series_id": series_id,
                    "api_key": self.api_key,
                    "file_type": "json",
                    "start_date": start_date,
                    "end_date": end_date,
                }

                response = requests.get(url, params=params)

                if response.status_code == 200:
                    response_data = response.json()
                    observations = response_data.get("observations", [])

                    if observations:
                        # Convert to pandas Series
                        dates = []
                        values = []

                        for obs in observations:
                            try:
                                date = pd.to_datetime(obs["date"])
                                value = (
                                    float(obs["value"])
                                    if obs["value"] != "."
                                    else np.nan
                                )
                                dates.append(date)
                                values.append(value)
                            except (ValueError, KeyError):
                                continue

                        if dates and values:
                            series_data = pd.Series(values, index=dates, name=series_id)
                            data[series_id] = series_data
                            print(
                                f"✓ Retrieved {len(series_data)} observations for {series_id}"
                            )
                        else:
                            print(f"✗ No valid data for {series_id}")
                    else:
                        print(f"✗ No observations found for {series_id}")
                else:
                    print(f"✗ Error fetching {series_id}: HTTP {response.status_code}")

            except Exception as e:
                print(f"✗ Error fetching {series_id}: {e}")

        return data

    def create_dataframe(self, data_dict):
        """Convert dictionary of series data to a pandas DataFrame."""
        if not data_dict:
            return pd.DataFrame()

        # Find the common date range
        all_dates = set()
        for series in data_dict.values():
            all_dates.update(series.index)

        # Create a complete date range
        if all_dates:
            date_range = pd.date_range(min(all_dates), max(all_dates), freq="D")
            df = pd.DataFrame(index=date_range)

            # Add each series
            for series_id, series_data in data_dict.items():
                df[series_id] = series_data

            df.index.name = "Date"
            return df

        return pd.DataFrame()

    def save_data(self, df, filename):
        """Save data to CSV file."""
        if df.empty:
            print("No data to save")
            return None

        filepath = os.path.join(OUTPUT_DIR, filename)
        df.to_csv(filepath)
        print(f"Data saved to {filepath}")
        return filepath

    def plot_economic_indicators(self, df, indicators_to_plot=None):
        """Create plots for economic indicators."""
        if df.empty:
            print("No data to plot")
            return

        if indicators_to_plot is None:
            indicators_to_plot = [col for col in df.columns if col in df.columns]

        if not indicators_to_plot:
            print("No indicators to plot")
            return

        # Set up the plotting style
        plt.style.use("default")
        sns.set_palette("husl")

        # Create subplots
        n_indicators = len(indicators_to_plot)
        fig, axes = plt.subplots(n_indicators, 1, figsize=(15, 4 * n_indicators))

        if n_indicators == 1:
            axes = [axes]

        for i, indicator in enumerate(indicators_to_plot):
            if indicator in df.columns:
                ax = axes[i]
                df[indicator].dropna().plot(ax=ax, linewidth=2)

                # Get series info for title
                info = self.get_series_info(indicator)
                title = f'{indicator} - {info["title"]}' if info else indicator
                ax.set_title(title)
                ax.set_ylabel("Value")
                ax.grid(True, alpha=0.3)

        plt.tight_layout()
        plot_path = os.path.join(PLOTS_DIR, "economic_indicators.png")
        plt.savefig(plot_path, dpi=300, bbox_inches="tight")
        plt.show()
        print(f"Plot saved to {plot_path}")

    def generate_summary_statistics(self, df):
        """Generate summary statistics for the economic data."""
        if df.empty:
            return pd.DataFrame()

        summary = df.describe()

        # Add additional statistics
        summary.loc["missing_values"] = df.isnull().sum()
        summary.loc["missing_percentage"] = (df.isnull().sum() / len(df)) * 100

        return summary

    def run_analysis(self, series_ids=None, start_date=None, end_date=None):
        """Run a complete analysis of economic indicators."""
        if series_ids is None:
            series_ids = list(self.indicators.values())

        print("=== FRED Economic Data Analysis v2 ===")
        print(f"API Key: {self.api_key[:8]}...")
        print(
            f"Date Range: {start_date or DEFAULT_START_DATE} to {end_date or DEFAULT_END_DATE}"
        )
        print(f"Series to analyze: {series_ids}")
        print("=" * 50)

        # Fetch data
        data = self.get_economic_data(series_ids, start_date, end_date)

        if not data:
            print("No data retrieved. Please check your API key and series IDs.")
            return None, None

        # Create DataFrame
        df = self.create_dataframe(data)

        if df.empty:
            print("No data to analyze")
            return None, None

        # Save data
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        self.save_data(df, f"fred_economic_data_{timestamp}.csv")

        # Generate summary statistics
        summary = self.generate_summary_statistics(df)
        print("\n=== Summary Statistics ===")
        print(summary)

        # Create plots
        print("\n=== Creating Visualizations ===")
        self.plot_economic_indicators(df)

        return df, summary


def main():
    """Main function to run the FRED data analysis."""
    collector = FREDDataCollectorV2()

    # Example: Analyze key economic indicators
    key_indicators = ["GDP", "UNRATE", "CPIAUCSL", "FEDFUNDS", "DGS10"]

    try:
        df, summary = collector.run_analysis(series_ids=key_indicators)

        if df is not None:
            print("\n=== Analysis Complete ===")
            print(f"Data shape: {df.shape}")
            print(f"Date range: {df.index.min()} to {df.index.max()}")
        else:
            print("\n=== Analysis Failed ===")

    except Exception as e:
        print(f"Error during analysis: {e}")


if __name__ == "__main__":
    main()