Cannabis Operations Suite

-- Core multi-tenant structure
CREATE TABLE organizations (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    name VARCHAR(255) NOT NULL,
    slug VARCHAR(100) UNIQUE NOT NULL,
    plan_tier VARCHAR(50) DEFAULT 'starter',
    subscription_status VARCHAR(50) DEFAULT 'trial',
    trial_ends_at TIMESTAMP,
    settings JSONB DEFAULT '{}',
    created_at TIMESTAMP DEFAULT NOW()
);

-- Row Level Security for multi-tenancy
ALTER TABLE organizations ENABLE ROW LEVEL SECURITY;

CREATE POLICY org_isolation_policy ON organizations
    FOR ALL TO authenticated_users
    USING (id = current_setting('app.current_org_id')::UUID);

-- Cannabis-specific product schema
CREATE TABLE products (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    organization_id UUID NOT NULL REFERENCES organizations(id),
    location_id UUID NOT NULL REFERENCES locations(id),
    sku VARCHAR(100),
    name VARCHAR(255) NOT NULL,
    category VARCHAR(100), -- flower, edibles, concentrates
    strain_name VARCHAR(255),
    strain_type VARCHAR(50), -- indica, sativa, hybrid
    thc_percentage DECIMAL(5,2),
    cbd_percentage DECIMAL(5,2),
    current_quantity INTEGER DEFAULT 0,
    reorder_point INTEGER,
    unit_cost DECIMAL(10,2),
    retail_price DECIMAL(10,2),
    expiration_date DATE,
    is_active BOOLEAN DEFAULT TRUE,
    created_at TIMESTAMP DEFAULT NOW()
);

from prophet import Prophet
import pandas as pd
from datetime import datetime, timedelta
import numpy as np

class CannabisInventoryForecaster:
    def __init__(self):
        self.models = {}
        
    def prepare_cannabis_data(self, sales_data: pd.DataFrame) -> pd.DataFrame:
        """Prepare sales data for Prophet with cannabis-specific features"""
        df = sales_data[['date', 'quantity_sold']].copy()
        df.columns = ['ds', 'y']
        
        # Ensure daily frequency
        df['ds'] = pd.to_datetime(df['ds'])
        df = df.groupby('ds')['y'].sum().reset_index()
        
        # Fill missing dates with 0 sales
        date_range = pd.date_range(
            start=df['ds'].min(), 
            end=df['ds'].max(), 
            freq='D'
        )
        full_df = pd.DataFrame({'ds': date_range})
        df = full_df.merge(df, on='ds', how='left').fillna(0)
        
        return df
    
    def create_cannabis_holidays(self) -> pd.DataFrame:
        """Create cannabis-specific holidays"""
        holidays = []
        
        # 4/20 Cannabis Holiday
        for year in range(2020, 2030):
            holidays.append({
                'holiday': '420_day',
                'ds': pd.to_datetime(f'{year}-04-20'),
                'lower_window': -2,
                'upper_window': 2,
            })
        
        # Green Wednesday (day before Thanksgiving)
        # Add other cannabis industry holidays
        
        return pd.DataFrame(holidays)
    
    async def forecast_product_demand(
        self, 
        product_id: str, 
        sales_history: pd.DataFrame,
        days_ahead: int = 14
    ) -> Dict[str, Any]:
        """Generate demand forecast for specific product"""
        
        # Prepare data
        df = self.prepare_cannabis_data(sales_history)
        
        if len(df) < 30:  # Minimum data requirement
            return {"error": "Insufficient data for forecasting"}
        
        # Initialize Prophet with cannabis-specific parameters
        model = Prophet(
            daily_seasonality=True,
            weekly_seasonality=True,
            yearly_seasonality=True,
            holidays=self.create_cannabis_holidays(),
            seasonality_mode='multiplicative',  # Better for sales data
            changepoint_prior_scale=0.05,  # Conservative for cannabis
            holidays_prior_scale=10.0  # Strong holiday effects
        )
        
        # Add custom seasonality for cannabis patterns
        model.add_seasonality(
            name='monthly_payday', 
            period=30.5, 
            fourier_order=5
        )
        
        # Fit model
        model.fit(df)
        
        # Generate forecast
        future = model.make_future_dataframe(periods=days_ahead)
        forecast = model.predict(future)
        
        # Calculate accuracy metrics if we have recent data
        accuracy_metrics = self._calculate_forecast_accuracy(
            df, forecast, days_back=7
        )
        
        # Cache model for this product
        self.models[product_id] = {
            'model': model,
            'last_updated': datetime.now(),
            'accuracy': accuracy_metrics
        }
        
        return {
            'forecast': forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']].tail(days_ahead).to_dict('records'),
            'accuracy_mape': accuracy_metrics.get('mape', 0),
            'confidence_score': self._calculate_confidence_score(forecast, df),
            'model_version': 'prophet_v1.1'
        }
    
    def _calculate_confidence_score(self, forecast: pd.DataFrame, historical: pd.DataFrame) -> float:
        """Calculate confidence score based on forecast uncertainty and historical variance"""
        recent_forecast = forecast.tail(14)
        uncertainty = (recent_forecast['yhat_upper'] - recent_forecast['yhat_lower']).mean()
        historical_variance = historical['y'].var()
        
        # Lower uncertainty relative to historical variance = higher confidence
        confidence = max(0.1, min(0.95, 1 - (uncertainty / (historical_variance + 1))))
        return round(confidence, 3)

import asyncio
from typing import Dict, List, Optional
import numpy as np
from dataclasses import dataclass

@dataclass
class PricingRule:
    min_margin_percent: float
    max_discount_percent: float
    competitor_weight: float
    inventory_weight: float
    demand_weight: float

class DynamicPricingEngine:
    def __init__(self):
        self.competitor_monitor = CompetitorMonitor()
        self.pricing_rules = {
            'flower': PricingRule(25.0, 15.0, 0.4, 0.35, 0.25),
            'edibles': PricingRule(35.0, 20.0, 0.3, 0.4, 0.3),
            'concentrates': PricingRule(40.0, 12.0, 0.5, 0.3, 0.2)
        }
    
    async def calculate_optimal_price(
        self, 
        product_id: str,
        current_price: float,
        unit_cost: float,
        current_inventory: int,
        demand_forecast: float,
        category: str
    ) -> Dict[str, Any]:
        """Calculate optimal price using multi-factor algorithm"""
        
        rule = self.pricing_rules.get(category, self.pricing_rules['flower'])
        
        # Get competitor pricing data
        competitor_prices = await self.competitor_monitor.get_competitor_prices(
            product_category=category,
            strain_type=product_id  # Simplified
        )
        
        if not competitor_prices:
            return {"error": "No competitor data available"}
        
        # Calculate pricing factors
        factors = await self._calculate_pricing_factors(
            competitor_prices, current_inventory, demand_forecast, rule
        )
        
        # Base price from competitor median
        base_price = np.median(competitor_prices)
        
        # Apply factors
        suggested_price = base_price * (
            1 + 
            factors['competitor_adjustment'] * rule.competitor_weight +
            factors['inventory_adjustment'] * rule.inventory_weight +
            factors['demand_adjustment'] * rule.demand_weight
        )
        
        # Ensure minimum margin
        min_price = unit_cost * (1 + rule.min_margin_percent / 100)
        max_discount_price = current_price * (1 - rule.max_discount_percent / 100)
        
        final_price = max(min_price, min(suggested_price, current_price * 1.1))
        
        return {
            'suggested_price': round(final_price, 2),
            'current_price': current_price,
            'price_change_percent': round(((final_price - current_price) / current_price) * 100, 2),
            'competitor_median': round(base_price, 2),
            'factors': factors,
            'confidence_score': self._calculate_pricing_confidence(factors),
            'reason': self._generate_pricing_reason(factors, final_price, current_price)
        }
    
    async def _calculate_pricing_factors(
        self, 
        competitor_prices: List[float], 
        inventory: int, 
        demand_forecast: float,
        rule: PricingRule
    ) -> Dict[str, float]:
        """Calculate individual pricing adjustment factors"""
        
        # Competitor factor: negative if we're above market, positive if below
        median_competitor = np.median(competitor_prices)
        competitor_factor = -0.05 if len(competitor_prices) > 3 else 0
        
        # Inventory factor: negative for overstock, positive for low stock
        days_supply = inventory / max(demand_forecast, 0.1)
        if days_supply > 30:  # Overstock
            inventory_factor = -0.08
        elif days_supply < 7:  # Low stock
            inventory_factor = 0.12
        else:
            inventory_factor = 0
        
        # Demand factor: positive for high demand, negative for low
        if demand_forecast > inventory * 0.1:  # High demand relative to stock
            demand_factor = 0.05
        elif demand_forecast < inventory * 0.02:  # Low demand
            demand_factor = -0.03
        else:
            demand_factor = 0
        
        return {
            'competitor_adjustment': competitor_factor,
            'inventory_adjustment': inventory_factor,
            'demand_adjustment': demand_factor
        }

class CompetitorMonitor:
    """Monitor competitor pricing from Weedmaps/Leafly"""
    
    async def get_competitor_prices(self, product_category: str, strain_type: str) -> List[float]:
        """Fetch competitor prices (implement based on your scraping strategy)"""
        # This would integrate with your web scraping or API solution
        # For now, return mock data
        return [25.50, 27.00, 24.75, 26.25, 25.00]

# Dockerfile for FastAPI Backend
FROM python:3.11-slim

WORKDIR /app

# Install system dependencies
RUN apt-get update && apt-get install -y \
    gcc \
    g++ \
    && rm -rf /var/lib/apt/lists/*

# Install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY . .

# Create non-root user for security
RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app
USER appuser

EXPOSE 8000

# Health check
HEALTHCHECK --interval=30s --timeout=30s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml for local development
version: '3.8'

services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/cannabis_db
      - REDIS_URL=redis://redis:6379
    depends_on:
      - db
      - redis
    volumes:
      - ./app:/app/app
  
  db:
    image: timescale/timescaledb:latest-pg14
    environment:
      POSTGRES_DB: cannabis_db
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data
  
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    command: redis-server --appendonly yes
    volumes:
      - redis_data:/data

volumes:
  postgres_data:
  redis_data: