Plot Inventory Data Over Time

import plotly.graph_objects as go
import ast
import pandas as pd
import numpy as np
import os
import logging
from typing import Tuple, Union, List, Optional, Dict

# Initialize logging
logging.basicConfig(level=logging.INFO)

def load_and_clean_data(file_path: str, asin: str) -> Union[pd.DataFrame, None]:
    """Load and clean the DataFrame."""
    if not os.path.exists(file_path):
        logging.error("The specified file does not exist.")
        return None

    try:
        df = pd.read_csv(file_path)
    except Exception as e:
        logging.error(f"An error occurred while reading the CSV file: {e}")
        return None

    df_filtered = df[df['asin'] == asin]
    if df_filtered.empty:
        logging.error(f"No data available for ASIN: {asin}")
        return None
    
    df_filtered['offers'] = df_filtered['offers'].apply(ast.literal_eval)
    df_filtered['time_collected'] = pd.to_datetime(df_filtered['time_collected'])
    
    return df_filtered

def calculate_prices(offers_list: List[Dict]) -> Tuple[Optional[float], Optional[float]]:
    prices = [offer['price_info']['price'] for offer in offers_list if 'price_info' in offer and 'price' in offer['price_info']]
    prime_prices = [offer['price_info']['price'] for offer in offers_list if offer.get('is_prime') and 'price_info' in offer and 'price' in offer['price_info']]
    
    avg_price = np.mean(prices) if prices else None
    lowest_prime_price = np.min(prime_prices) if prime_prices else None
    
    return avg_price, lowest_prime_price

def plot_fig(df_filtered: pd.DataFrame, asin: str) -> go.Figure:
    fig = go.Figure()
    
    fig.add_trace(go.Scatter(x=df_filtered['time_collected'], y=df_filtered['total_inventory'],
                             mode='lines+markers',
                             name='Total Inventory',
                             line=dict(color='blue')))
    
    fig.add_trace(go.Scatter(x=df_filtered['time_collected'], y=df_filtered['avg_price'],
                             mode='lines+markers',
                             name='Average Price',
                             yaxis="y2",
                             line=dict(color='red', dash='dash')))
    
    fig.add_trace(go.Scatter(x=df_filtered['time_collected'], y=df_filtered['lowest_prime_price'],
                             mode='markers',
                             name='Lowest Prime Price',
                             yaxis="y2",
                             marker=dict(color='green', symbol='square')))
    
    fig.update_layout(
        title=f'Inventory and Pricing Over Time for ASIN: {asin}',
        xaxis=dict(title='Time Collected'),
        yaxis=dict(title='Total Inventory', color='black'),
        yaxis2=dict(title='Price ($)', overlaying='y', side='right', color='red'),
    )
    
    return fig

file_path = 'sample_data.csv' # replace with your file path
asin = 'B01N1LL62W' # replace with your desired ASIN

df_filtered = load_and_clean_data(file_path, asin)

if df_filtered is None:
    logging.error("Could not proceed with the visualization.")
    exit(1)

df_filtered['avg_price'], df_filtered['lowest_prime_price'] = zip(*df_filtered['offers'].apply(calculate_prices))

fig = plot_fig(df_filtered, asin)
fig.show()