PHANTOM/experiments/procesing/pipelines.py

from sklearn.pipeline import Pipeline
import pandas as pd
from procesing.context import PipelineContext
from procesing.providers import SupabaseProvider, BackendAPIProvider
from typing import Union
from procesing.steps import (
    FetchInteractionsStep,
    FetchPriceLogsStep,
    FetchExperimentsStep,
    JoinExperimentsStep,
    CreatePriceBucketsStep,
    AugmentEventNamesStep,
    ChunkByTimeWindowStep,
    ComputeDemandForChunksStep,
    AggregatePriceLogsStep,
    ComputeElasticityStep,
    # BuildStateSpaceStep,
    FitPricingFunctionStep,
    PredictPricesStep,
)

def interaction_extraction_pipeline(context: PipelineContext):
    """Pipeline for extracting and augmenting interaction data"""
    return Pipeline([
        ('fetch', FetchInteractionsStep(context)),
        ('create_buckets', CreatePriceBucketsStep(context)),
        ('augment_events', AugmentEventNamesStep(context)),
    ])


def price_extraction_pipeline(context: PipelineContext):
    """Pipeline for extracting price logs"""
    return Pipeline([
        ('fetch', FetchPriceLogsStep(context)),
    ])


def elasticity_computation_pipeline(context: PipelineContext,
                                    interactions_df: pd.DataFrame,
                                    price_logs_df: pd.DataFrame):
    """
    Compute elasticity from interactions and price logs.
    Manual orchestration needed for branching logic.
    """
    # branch 1: chunk interactions and compute demand
    chunk_step = ChunkByTimeWindowStep(context)
    interaction_chunks = chunk_step.transform(interactions_df)

    demand_step = ComputeDemandForChunksStep(context)
    demand_chunks = demand_step.transform(interaction_chunks)

    # branch 2: aggregate price logs
    price_step = AggregatePriceLogsStep(context)
    price_chunks = price_step.transform(price_logs_df)

    # convergence: compute elasticity
    elasticity_step = ComputeElasticityStep(context)
    elasticity_df = elasticity_step.transform((demand_chunks, price_chunks))

    return elasticity_df


def pricing_pipeline(context: PipelineContext, elasticity_df: pd.DataFrame):
    """
    Generate optimal prices from elasticity estimates.
    """
    # build state space
    state_step = BuildStateSpaceStep(context)
    state_space = state_step.transform(elasticity_df)

    # fit pricing function
    fit_step = FitPricingFunctionStep(context)
    pricer = fit_step.transform(elasticity_df)

    # predict prices
    predict_step = PredictPricesStep(context)
    prices_df = predict_step.transform((pricer, state_space))

    return prices_df


def full_pipeline(context: PipelineContext):
    """
    Complete end-to-end pipeline: data extraction -> elasticity -> pricing
    Returns: (elasticity_df, prices_df)
    """
    # extract interactions
    interaction_pipe = interaction_extraction_pipeline(context)
    interactions_df = interaction_pipe.fit_transform(None)

    # extract price logs
    price_pipe = price_extraction_pipeline(context)
    price_logs_df = price_pipe.fit_transform(None)

    if interactions_df.empty or price_logs_df.empty:
        return None, None

    # compute elasticity
    elasticity_df = elasticity_computation_pipeline(
        context,
        interactions_df,
        price_logs_df
    )

    if elasticity_df is None or elasticity_df.empty:
        return elasticity_df, None

    # generate prices
    prices_df = pricing_pipeline(context, elasticity_df)

    return elasticity_df, prices_df


if __name__ == '__main__':

    class Provider(SupabaseProvider, BackendAPIProvider):
        def __init__(self, backend_url: str):
            SupabaseProvider.__init__(self)
            BackendAPIProvider.__init__(self, backend_url=backend_url)
    # example run
    context = PipelineContext(
        provider=Provider(backend_url="http://localhost:5000"),
        store_mode='hotel',
        # 15 min not month
        window_size='15min',
    )

    elasticity_df, prices_df = full_pipeline(context)

    if elasticity_df is not None and not elasticity_df.empty:
        print("Elasticity Estimates:")
        print(elasticity_df.to_string(index=False))
    else:
        print("No elasticity estimates computed.")

    if prices_df is not None and not prices_df.empty:
        print("\nPredicted Prices:")
        print(prices_df.to_string(index=False))
    else:
        print("No prices predicted.")