Code Editor for app.py

import streamlit as st
import pandas as pd
import plotly.express as px
from sqlalchemy import create_engine, Column, String, Float, Integer
from sqlalchemy.orm import DeclarativeBase, Session

# Database setup
class Base(DeclarativeBase):
    pass

class Product(Base):
    __tablename__ = 'products'
    
    id = Column(Integer, primary_key=True)
    hs_code = Column(String)
    product_name = Column(String)
    exporting_country = Column(String)
    importing_country = Column(String)
    tariff_rate = Column(Float)
    unit_price = Column(Float)
    supplier_name = Column(String)
    risk_score = Column(Float)

# Create engine and tables
engine = create_engine('sqlite:///supply_chain.db')
Base.metadata.create_all(bind=engine)

def load_sample_data():
    """Load sample data if database is empty"""
    with Session(engine) as session:
        if session.query(Product).first() is None:
            sample_data = [
                Product(
                    hs_code='8471.30',
                    product_name='Laptops',
                    exporting_country='China',
                    importing_country='USA',
                    tariff_rate=25.0,
                    unit_price=500.0,
                    supplier_name='TechCorp',
                    risk_score=0.8
                ),
                Product(
                    hs_code='8471.30',
                    product_name='Laptops',
                    exporting_country='Vietnam',
                    importing_country='USA',
                    tariff_rate=0.0,
                    unit_price=550.0,
                    supplier_name='VietTech',
                    risk_score=0.4
                ),
                # Add more sample data as needed
            ]
            session.add_all(sample_data)
            session.commit()

# Load sample data
load_sample_data()

st.title('Supply Chain Tariff Impact Analyzer')

# Sidebar for filters
st.sidebar.header('Filters')
with Session(engine) as session:
    hs_codes = [p.hs_code for p in session.query(Product.hs_code).distinct()]
    countries = [p.exporting_country for p in session.query(Product.exporting_country).distinct()]

selected_hs = st.sidebar.selectbox('Select HS Code', hs_codes)
selected_country = st.sidebar.selectbox('Select Exporting Country', countries)

# Main content
st.header('Product Analysis')

# Query data
with Session(engine) as session:
    products = session.query(Product).filter(
        Product.hs_code == selected_hs,
        Product.exporting_country == selected_country
    ).all()
    
    # Convert to DataFrame for display
    df = pd.DataFrame([{
        'HS Code': p.hs_code,
        'Product': p.product_name,
        'Exporter': p.exporting_country,
        'Importer': p.importing_country,
        'Tariff (%)': p.tariff_rate,
        'Unit Price': p.unit_price,
        'Supplier': p.supplier_name,
        'Risk Score': p.risk_score
    } for p in products])

if not df.empty:
    # Display current supplier info
    st.subheader('Current Supplier Information')
    st.dataframe(df)
    
    # Calculate impact
    total_cost = df['Unit Price'].mean() * (1 + df['Tariff (%)'].mean()/100)
    st.metric('Total Cost per Unit (including tariffs)', f'${total_cost:.2f}')
    
    # Alternative suppliers
    st.subheader('Alternative Suppliers Analysis')
    with Session(engine) as session:
        alt_suppliers = session.query(Product).filter(
            Product.hs_code == selected_hs,
            Product.exporting_country != selected_country
        ).all()
        
        alt_df = pd.DataFrame([{
            'Exporter': p.exporting_country,
            'Supplier': p.supplier_name,
            'Tariff (%)': p.tariff_rate,
            'Unit Price': p.unit_price,
            'Risk Score': p.risk_score,
            'Total Cost': p.unit_price * (1 + p.tariff_rate/100)
        } for p in alt_suppliers])
    
    if not alt_df.empty:
        # Plot alternative suppliers comparison
        fig = px.scatter(alt_df, 
                        x='Total Cost', 
                        y='Risk Score',
                        size='Tariff (%)',
                        hover_data=['Supplier', 'Exporter'],
                        title='Alternative Suppliers: Cost vs Risk Analysis')
        st.plotly_chart(fig)
        
        st.dataframe(alt_df.sort_values('Total Cost'))
    else:
        st.warning('No alternative suppliers found for this product.')
    
    # Add new supplier form
    st.subheader('Add New Supplier')
    with st.form('new_supplier'):
        new_hs = st.text_input('HS Code', selected_hs)
        new_product = st.text_input('Product Name')
        new_exporter = st.text_input('Exporting Country')
        new_importer = st.text_input('Importing Country')
        new_tariff = st.number_input('Tariff Rate (%)', min_value=0.0)
        new_price = st.number_input('Unit Price ($)', min_value=0.0)
        new_supplier = st.text_input('Supplier Name')
        new_risk = st.slider('Risk Score', 0.0, 1.0, 0.5)
        
        if st.form_submit_button('Add Supplier'):
            with Session(engine) as session:
                new_product = Product(
                    hs_code=new_hs,
                    product_name=new_product,
                    exporting_country=new_exporter,
                    importing_country=new_importer,
                    tariff_rate=new_tariff,
                    unit_price=new_price,
                    supplier_name=new_supplier,
                    risk_score=new_risk
                )
                session.add(new_product)
                session.commit()
                st.success('New supplier added successfully!')
                st.rerun()
else:
    st.warning('No data available for the selected filters.')