pip install yfinance pandas backoff matplotlib seaborn

import datetime
import time
import logging
import os
from collections import deque

import pandas as pd
import yfinance as yf
import backoff
import matplotlib.pyplot as plt
import seaborn as sns

def create_state() -> dict:
    """
    Initializes the application state dictionary, including a logger.

    Returns
    -------
    dict
        An initial state dictionary containing a configured logger.
    """
    state = {}
    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
    state['logger'] = logging.getLogger(__name__)
    state['logger'].info("Application state initialized and logger configured.")
    return state

@backoff.on_exception(backoff.expo, (Exception, ConnectionError, TimeoutError), max_tries=5, factor=2, jitter=backoff.full_jitter)
def fetch_historical_data(state: dict, ticker: str, start_date: str, end_date: str) -> dict:
    """
    Fetches historical trading data for a given ticker using yfinance with exponential backoff.

    Parameters
    ----------
    state : dict
        Current state dictionary with a logger.
    ticker : str
        The stock ticker symbol (e.g., 'AAPL').
    start_date : str
        The start date for data fetching in 'YYYY-MM-DD' format.
    end_date : str
        The end date for data fetching in 'YYYY-MM-DD' format.

    Returns
    -------
    dict
        Updated state with fetched data (pandas DataFrame) or an empty DataFrame if failed.
    """
    logger = state['logger']
    logger.info(f"Attempting to fetch data for {ticker} from {start_date} to {end_date}")
    try:
        data = yf.download(ticker, start=start_date, end=end_date)
        if not data.empty:
            logger.info(f"Successfully fetched {len(data)} rows for {ticker}.")
            state[f'{ticker}_data'] = data
        else:
            logger.warning(f"No data fetched for {ticker} between {start_date} and {end_date}.")
            state[f'{ticker}_data'] = pd.DataFrame()
    except Exception as e:
        logger.error(f"Failed to fetch data for {ticker} after retries: {e}")
        state[f'{ticker}_data'] = pd.DataFrame() # Ensure key exists even on failure
    return state

def store_data_local(state: dict, data_df: pd.DataFrame, file_path: str) -> dict:
    """
    Stores a pandas DataFrame to a local CSV file.

    Parameters
    ----------
    state : dict
        Current state dictionary with a logger.
    data_df : pd.DataFrame
        The DataFrame containing the historical data to be stored.
    file_path : str
        The full path including the filename where the data should be saved.

    Returns
    -------
    dict
        Updated state dictionary.
    """
    logger = state['logger']
    try:
        os.makedirs(os.path.dirname(file_path), exist_ok=True)
        data_df.to_csv(file_path, index=True)
        logger.info(f"Data successfully stored to {file_path}")
    except Exception as e:
        logger.error(f"Failed to store data to {file_path}: {e}")
    return state

def load_data_local(state: dict, file_path: str, ticker: str) -> dict:
    """
    Loads a pandas DataFrame from a local CSV file.

    Parameters
    ----------
    state : dict
        Current state dictionary with a logger.
    file_path : str
        The full path to the CSV file to be loaded.
    ticker : str
        The ticker symbol to name the key in the state.

    Returns
    -------
    dict
        Updated state with loaded data (pandas DataFrame) or an empty DataFrame if failed.
    """
    logger = state['logger']
    try:
        # Read with multi-level header and set the first column (Date) as index.
        # This handles the specific format yfinance produces in some environments.
        loaded_df = pd.read_csv(file_path, header=[0, 1], index_col=0, parse_dates=True)

        # Drop the second level of the MultiIndex (e.g., 'BTC-USD')
        loaded_df.columns = loaded_df.columns.droplevel(1)

        # The first column is now 'Price' (from the first header row) which is a placeholder.
        # We need to drop this column to get only the data columns.
        # Ensure we only drop if 'Price' column exists and is not a data column.
        if 'Price' in loaded_df.columns:
            loaded_df = loaded_df.drop(columns=['Price'])

        # Convert relevant columns to numeric, coercing errors to NaN
        numeric_cols = ['Close', 'High', 'Low', 'Open', 'Volume', 'Adj Close'] # Include Adj Close if present
        for col in numeric_cols:
            if col in loaded_df.columns:
                loaded_df[col] = pd.to_numeric(loaded_df[col], errors='coerce')

        logger.info(f"Data successfully loaded from {file_path}. Shape: {loaded_df.shape}")
        state[f'{ticker}_loaded_data'] = loaded_df
    except FileNotFoundError:
        logger.warning(f"File not found at {file_path}. Returning empty DataFrame.")
        state[f'{ticker}_loaded_data'] = pd.DataFrame()
    except Exception as e:
        logger.error(f"Failed to load data from {file_path}: {e}")
        state[f'{ticker}_loaded_data'] = pd.DataFrame()
    return state

def summarize_historical_data(state: dict, data_df: pd.DataFrame, ticker: str) -> dict:
    """
    Calculates and displays summary statistics for the historical data.

    Parameters
    ----------
    state : dict
        Current state dictionary with a logger.
    data_df : pd.DataFrame
        The DataFrame containing the historical data to be summarized.
    ticker : str
        The ticker symbol for the data being summarized.

    Returns
    -------
    dict
        Updated state dictionary.
    """
    logger = state['logger']
    if data_df.empty:
        logger.warning(f"No data to summarize for {ticker}. DataFrame is empty.")
        return state

    logger.info(f"Displaying summary for {ticker}:")
    print(f"\n--- Summary for {ticker} ---")
    print("Shape:", data_df.shape)
    print("\nFirst 5 rows:")
    display(data_df.head())
    print("\nLast 5 rows:")
    display(data_df.tail())
    print("\nDescriptive Statistics:")
    display(data_df.describe())
    print("\nMissing values:")
    display(data_df.isnull().sum().to_frame(name='Missing Count'))

    state[f'{ticker}_summary'] = data_df.describe().to_dict()
    return state

# Initialize the state
app_state = create_state()
logger = app_state['logger']

# Define parameters for demonstration
tickers = ['BTC-USD', 'ETH-USD']
start_date = '2023-01-01'
end_date = '2025-01-01'
archive_dir = 'archived_data'

# Ensure the archive directory exists
os.makedirs(archive_dir, exist_ok=True)

# --- Step 1: Fetch and store data for each ticker ---
for ticker in tickers:
    logger.info(f"Processing ticker: {ticker}")
    app_state = fetch_historical_data(app_state, ticker, start_date, end_date)

    fetched_data = app_state.get(f'{ticker}_data')
    if fetched_data is not None and not fetched_data.empty:
        file_path = os.path.join(archive_dir, f'{ticker}.csv')
        app_state = store_data_local(app_state, fetched_data, file_path)
    else:
        logger.warning(f"No data to store for {ticker}.")

# --- Step 2: Load data from local storage and summarize ---
all_loaded_data = {}
for ticker in tickers:
    file_path = os.path.join(archive_dir, f'{ticker}.csv')
    app_state = load_data_local(app_state, file_path, ticker)
    loaded_data_df = app_state.get(f'{ticker}_loaded_data')

    if loaded_data_df is not None and not loaded_data_df.empty:
        all_loaded_data[ticker] = loaded_data_df
        app_state = summarize_historical_data(app_state, loaded_data_df, ticker)
    else:
        logger.error(f"Could not load data for {ticker} from {file_path}.")

if all_loaded_data:
    for ticker, df in all_loaded_data.items():
        if not df.empty:
            plt.figure(figsize=(14, 7))
            plt.plot(df.index, df['Close'], label=f'{ticker} Close Price') # Access 'Close' directly

            plt.title(f'Historical Closing Prices for {ticker} ({start_date} to {end_date})') # Dynamic title
            plt.xlabel('Date')
            plt.ylabel('Close Price (USD)')
            plt.legend()
            plt.grid(True)
            plt.tight_layout()
            plt.show()
else:
    logger.warning("No data available to plot historical closing prices.")

if all_loaded_data:
    fig, axes = plt.subplots(len(tickers), 2, figsize=(15, 5 * len(tickers)))
    fig.suptitle('Daily Returns Distribution and Rolling Average', y=1.02)

    for i, (ticker, df_original) in enumerate(all_loaded_data.items()):
        if not df_original.empty:
            # Make an explicit copy to avoid SettingWithCopyWarning
            df = df_original.copy()

            # Access 'Close' directly for daily return calculation
            df['Daily Return'] = df['Close'].pct_change(fill_method=None) * 100 # Added fill_method=None
            df = df.dropna()

            # Plotting Daily Returns Distribution
            sns.histplot(df['Daily Return'], kde=True, ax=axes[i, 0], bins=50)
            axes[i, 0].set_title(f'{ticker} Daily Returns Distribution')
            axes[i, 0].set_xlabel('Daily Return (%)')
            axes[i, 0].set_ylabel('Frequency')
            axes[i, 0].grid(True, linestyle='--', alpha=0.7)

            # Demonstrate deque for rolling window of average returns
            window_size = 30 # 30-day rolling window
            returns_window = deque(maxlen=window_size)
            rolling_averages = []

            for ret in df['Daily Return']:
                returns_window.append(ret)
                if len(returns_window) == window_size:
                    rolling_averages.append(sum(returns_window) / window_size)
                else:
                    rolling_averages.append(None) # Pad with None until window is full

            df['Rolling Avg Return'] = rolling_averages

            # Update the original dictionary with the modified DataFrame
            all_loaded_data[ticker] = df

            # Plotting Rolling Average Daily Returns
            axes[i, 1].plot(df.index, df['Rolling Avg Return'], label=f'{ticker} {window_size}-Day Rolling Avg Return', color='orange')
            axes[i, 1].set_title(f'{ticker} {window_size}-Day Rolling Average Daily Returns')
            axes[i, 1].set_xlabel('Date')
            axes[i, 1].set_ylabel('Rolling Avg Return (%)')
            axes[i, 1].legend()
            axes[i, 1].grid(True, linestyle='--', alpha=0.7)

    plt.tight_layout(rect=[0, 0.03, 1, 0.98])
    plt.show()
else:
    logger.warning("No data available to plot daily returns distribution.")