chore: refactoring function definition to avoid reinit

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

README.md

@@ -1,8 +1,12 @@
+<img width="200" align="left" src="https://github.com/user-attachments/assets/d148b00d-e9f9-4280-89cc-0cc866e17251" />
 
-<img width="1952" height="2176" alt="nobody_knows" src="https://github.com/user-attachments/assets/d148b00d-e9f9-4280-89cc-0cc866e17251" />
+### PHANTOM
 
 [![Build PDF](https://github.com/velocitatem/PHANTOM/actions/workflows/latex.yml/badge.svg)](https://github.com/velocitatem/PHANTOM/actions/workflows/latex.yml)
+[![TPU Research Cloud](https://img.shields.io/badge/TPU%20Research%20Cloud-TRC%20supported-4285F4?logo=googlecloud&logoColor=white)](https://sites.research.google/trc/faq/)
+[![Vercel Deploy](https://deploy-badge.vercel.app/?url=https://phantom-hotel.vercel.app&name=Hotel)](https://phantom-hotel.vercel.app)
+[![Vercel Deploy](https://deploy-badge.vercel.app/?url=https://phantom-airline.vercel.app&name=Airline)](https://phantom-airline.vercel.app)
+
+
 
-- https://phantom-hotel.vercel.app/
-- https://phantom-airline.vercel.app/
 

docker-compose.yml

@@ -103,12 +103,6 @@ services:
       - _AIRFLOW_WWW_USER_PASSWORD=admin
       - REDIS_HOST=redis
       - REDIS_PORT=6379
-    volumes:
-      - ./experiments/airflow/dags:/opt/airflow/dags
-      - ./experiments/airflow/logs:/opt/airflow/logs
-      - ./experiments/airflow/plugins:/opt/airflow/plugins
-      - ./experiments/procesing:/opt/airflow/procesing
-      - ./lib:/opt/airflow/lib
     command: version
     restart: "no"
 
@@ -129,6 +123,7 @@ services:
       - AIRFLOW__CORE__LOAD_EXAMPLES=false
       - AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
       - AIRFLOW__WEBSERVER__EXPOSE_CONFIG=true
+      - AIRFLOW__WEBSERVER__SECRET_KEY=${AIRFLOW_SECRET_KEY}
       - KAFKA_HOST=kafka
       - KAFKA_PORT=29092
       - BACKEND_URL=http://backend:5000
@@ -138,12 +133,6 @@ services:
       - REDIS_PORT=6379
     ports:
       - "${AIRFLOW_WEBSERVER_PORT:-8085}:8080"
-    volumes:
-      - ./experiments/airflow/dags:/opt/airflow/dags:ro
-      - ./experiments/airflow/logs:/opt/airflow/logs
-      - ./experiments/airflow/plugins:/opt/airflow/plugins:ro
-      - ./experiments/procesing:/opt/airflow/procesing:ro
-      - ./lib:/opt/airflow/lib:ro
     command: webserver
     restart: unless-stopped
     healthcheck:
@@ -170,6 +159,7 @@ services:
       - AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION=true
       - AIRFLOW__CORE__LOAD_EXAMPLES=false
       - AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
+      - AIRFLOW__WEBSERVER__SECRET_KEY=${AIRFLOW_SECRET_KEY}
       - KAFKA_HOST=kafka
       - KAFKA_PORT=29092
       - BACKEND_URL=http://backend:5000
@@ -177,12 +167,6 @@ services:
       - NEXT_PUBLIC_SUPABASE_ANON_KEY=${NEXT_PUBLIC_SUPABASE_ANON_KEY}
       - REDIS_HOST=redis
       - REDIS_PORT=6379
-    volumes:
-      - ./experiments/airflow/dags:/opt/airflow/dags:ro
-      - ./experiments/airflow/logs:/opt/airflow/logs
-      - ./experiments/airflow/plugins:/opt/airflow/plugins:ro
-      - ./experiments/procesing:/opt/airflow/procesing:ro
-      - ./lib:/opt/airflow/lib:ro
     command: scheduler
     restart: unless-stopped
     healthcheck:

docker/Airflow.dockerfile

@@ -21,3 +21,10 @@ RUN pip install --no-cache-dir \
 
 # set airflow home
 ENV AIRFLOW_HOME=/opt/airflow
+
+COPY --chown=airflow:root experiments/airflow/dags ${AIRFLOW_HOME}/dags
+COPY --chown=airflow:root experiments/procesing ${AIRFLOW_HOME}/procesing
+COPY --chown=airflow:root lib ${AIRFLOW_HOME}/lib
+
+# create logs and plugins dirs (airflow expects them)
+RUN mkdir -p ${AIRFLOW_HOME}/logs ${AIRFLOW_HOME}/plugins

docker/Airflow.railway.dockerfile

@@ -0,0 +1,41 @@
+FROM apache/airflow:2.7.3-python3.11
+
+USER root
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    supervisor \
+    && apt-get clean \
+    && rm -rf /var/lib/apt/lists/*
+
+USER airflow
+
+COPY requirements.txt /tmp/requirements.txt
+RUN pip install --no-cache-dir -r /tmp/requirements.txt
+
+RUN pip install --no-cache-dir \
+    psycopg2-binary \
+    apache-airflow-providers-postgres
+
+ENV AIRFLOW_HOME=/opt/airflow
+ENV AIRFLOW__CORE__EXECUTOR=SequentialExecutor
+ENV AIRFLOW__CORE__LOAD_EXAMPLES=false
+ENV AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
+ENV AIRFLOW__WEBSERVER__EXPOSE_CONFIG=true
+
+# copy all code into image (standalone - no volume mounts needed)
+COPY --chown=airflow:root experiments/airflow/dags ${AIRFLOW_HOME}/dags
+COPY --chown=airflow:root experiments/procesing ${AIRFLOW_HOME}/procesing
+COPY --chown=airflow:root lib ${AIRFLOW_HOME}/lib
+
+RUN mkdir -p ${AIRFLOW_HOME}/logs ${AIRFLOW_HOME}/plugins
+
+# copy entrypoint script
+COPY --chown=airflow:root docker/airflow-railway-entrypoint.sh /entrypoint.sh
+USER root
+RUN chmod +x /entrypoint.sh
+USER airflow
+
+EXPOSE 8080
+
+ENTRYPOINT ["/entrypoint.sh"]

docker/airflow-railway-entrypoint.sh

@@ -0,0 +1,20 @@
+#!/bin/bash
+set -e
+
+# init db and create admin user on first run
+airflow db migrate
+
+# create admin user if not exists
+airflow users create \
+    --username "${AIRFLOW_ADMIN_USER:-admin}" \
+    --password "${AIRFLOW_ADMIN_PASSWORD:-admin}" \
+    --firstname Admin \
+    --lastname User \
+    --role Admin \
+    --email admin@example.com || true
+
+# start scheduler in background
+airflow scheduler &
+
+# start webserver in foreground (Railway needs one foreground process)
+exec airflow webserver --port ${PORT:-8080}

experiments/airflow/dags/surge_pricing_factory.py

@@ -0,0 +1,210 @@
+from airflow import DAG
+from airflow.operators.python import PythonOperator
+from airflow.utils.dates import days_ago
+from datetime import timedelta
+import pandas as pd
+import logging
+import sys
+import pickle
+
+sys.path.insert(0, '/opt/airflow')
+
+from procesing.context import PipelineContext
+from procesing.providers import SupabaseProvider, BackendAPIProvider
+from procesing.steps import (
+    FetchInteractionsStep,
+    FetchPriceLogsStep,
+    ComputeDemandStep,
+    AggregatePriceLogsStep,
+    JoinProductFeaturesStep,
+)
+from procesing.pricers.simple import SimpleSurgePricer
+
+DEFAULT_ARGS = {
+    'owner': 'phantom-research',
+    'depends_on_past': False,
+    'email_on_failure': False,
+    'email_on_retry': False,
+    'retries': 2,
+    'retry_delay': timedelta(minutes=5),
+}
+
+class CompositeProvider(SupabaseProvider, BackendAPIProvider):
+    def __init__(self):
+        SupabaseProvider.__init__(self)
+        BackendAPIProvider.__init__(self)
+
+def _get_provider():
+    return CompositeProvider()
+
+def _make_task_callables(store_mode: str):
+    """Generate task callables bound to a specific store_mode."""
+
+    def get_context(**kwargs):
+        return PipelineContext(provider=_get_provider(), store_mode=store_mode)
+
+    def fetch_interactions(**kwargs):
+        ctx = get_context(**kwargs)
+        df = FetchInteractionsStep(ctx).transform(None)
+        kwargs['ti'].xcom_push(key='interactions_raw', value=pickle.dumps(df))
+        logging.info(f"[{store_mode}] Fetched {len(df)} interaction records")
+        return len(df)
+
+    def fetch_price_logs(**kwargs):
+        ctx = get_context(**kwargs)
+        df = FetchPriceLogsStep(ctx).transform(None)
+        kwargs['ti'].xcom_push(key='price_logs_raw', value=pickle.dumps(df))
+        logging.info(f"[{store_mode}] Fetched {len(df)} price records")
+        return len(df)
+
+    def compute_demand(**kwargs):
+        ti = kwargs['ti']
+        df = pickle.loads(ti.xcom_pull(key='interactions_raw'))
+        ctx = get_context(**kwargs)
+        demand_df = ComputeDemandStep(ctx).transform(df)
+        ti.xcom_push(key='demand_data', value=pickle.dumps(demand_df))
+        logging.info(f"[{store_mode}] Computed demand for {len(demand_df)} products")
+        return len(demand_df)
+
+    def aggregate_price_logs(**kwargs):
+        ti = kwargs['ti']
+        df = pickle.loads(ti.xcom_pull(key='price_logs_raw'))
+        ctx = get_context(**kwargs)
+        price_df = AggregatePriceLogsStep(ctx).transform(df)
+        ti.xcom_push(key='price_data', value=pickle.dumps(price_df))
+        logging.info(f"[{store_mode}] Aggregated price logs for {len(price_df)} products")
+        return len(price_df)
+
+    def join_product_features(**kwargs):
+        ti = kwargs['ti']
+        demand_df = pickle.loads(ti.xcom_pull(key='demand_data'))
+        price_df = pickle.loads(ti.xcom_pull(key='price_data'))
+        ctx = get_context(**kwargs)
+        joined_df = JoinProductFeaturesStep(ctx).transform((demand_df, price_df))
+        ti.xcom_push(key='product_features', value=pickle.dumps(joined_df))
+        logging.info(f"[{store_mode}] Joined features for {len(joined_df)} products")
+        return len(joined_df)
+
+    def apply_surge_pricing(**kwargs):
+        ti = kwargs['ti']
+        product_features = pickle.loads(ti.xcom_pull(key='product_features'))
+        dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
+
+        data = product_features.rename(columns={'demand_score': 'demand'})
+        surge_pricer = SimpleSurgePricer(
+            high_threshold=dag_conf.get('high_threshold', 10),
+            low_threshold=dag_conf.get('low_threshold', 2),
+            surge_multiplier=dag_conf.get('surge_multiplier', 1.2),
+            discount_multiplier=dag_conf.get('discount_multiplier', 0.9)
+        )
+        surge_pricer.fit(data)
+        data['optimal_price'] = surge_pricer.predict()
+
+        prices_df = data[['productId', 'price', 'base_price', 'optimal_price', 'demand']].rename(columns={
+            'price': 'current_price', 'demand': 'demand_score'
+        })
+        ti.xcom_push(key='predicted_prices', value=pickle.dumps(prices_df))
+        logging.info(f"[{store_mode}] Applied surge pricing for {len(prices_df)} products")
+        return len(prices_df)
+
+    def publish_results(**kwargs):
+        ti = kwargs['ti']
+        prices_df = pickle.loads(ti.xcom_pull(key='predicted_prices'))
+        from lib.model_registry import ModelRegistry
+
+        registry = ModelRegistry()
+        dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
+
+        metadata = {
+            'timestamp': pd.Timestamp.now().isoformat(),
+            'store_mode': store_mode,
+            'dag_run_id': kwargs['dag_run'].run_id if kwargs.get('dag_run') else 'manual',
+            'pricing_method': 'surge',
+            'high_threshold': dag_conf.get('high_threshold', 10),
+            'low_threshold': dag_conf.get('low_threshold', 2),
+            'surge_multiplier': dag_conf.get('surge_multiplier', 1.2),
+            'discount_multiplier': dag_conf.get('discount_multiplier', 0.9)
+        }
+        registry.publish_prices(prices_df, model_name=f'{store_mode}_latest', metadata=metadata)
+        logging.info(f"[{store_mode}] Published surge pricing for {len(prices_df)} products")
+
+        return {
+            'n_products': len(prices_df),
+            'registry_status': 'success',
+            'store_mode': store_mode,
+            'mean_demand': float(prices_df['demand_score'].mean()) if 'demand_score' in prices_df.columns else None
+        }
+
+    return {
+        'fetch_interactions': fetch_interactions,
+        'fetch_price_logs': fetch_price_logs,
+        'compute_demand': compute_demand,
+        'aggregate_price_logs': aggregate_price_logs,
+        'join_product_features': join_product_features,
+        'apply_surge_pricing': apply_surge_pricing,
+        'publish_results': publish_results,
+    }
+
+
+def create_surge_pricing_dag(store_mode: str) -> DAG:
+    """Factory: generates a surge pricing DAG for a given store_mode."""
+    callables = _make_task_callables(store_mode)
+
+    dag = DAG(
+        f'surge_pricing_{store_mode}',
+        default_args=DEFAULT_ARGS,
+        description=f'Surge pricing pipeline for {store_mode} store mode',
+        schedule_interval='*/15 * * * *',
+        start_date=days_ago(1),
+        catchup=False,
+        max_active_runs=1,
+        tags=['pricing', 'surge', 'research', store_mode],
+    )
+
+    with dag:
+        t_fetch_interactions = PythonOperator(
+            task_id='fetch_interactions',
+            python_callable=callables['fetch_interactions'],
+            provide_context=True,
+        )
+        t_fetch_price_logs = PythonOperator(
+            task_id='fetch_price_logs',
+            python_callable=callables['fetch_price_logs'],
+            provide_context=True,
+        )
+        t_compute_demand = PythonOperator(
+            task_id='compute_demand',
+            python_callable=callables['compute_demand'],
+            provide_context=True,
+        )
+        t_aggregate_prices = PythonOperator(
+            task_id='aggregate_price_logs',
+            python_callable=callables['aggregate_price_logs'],
+            provide_context=True,
+        )
+        t_join_features = PythonOperator(
+            task_id='join_product_features',
+            python_callable=callables['join_product_features'],
+            provide_context=True,
+        )
+        t_surge_pricing = PythonOperator(
+            task_id='apply_surge_pricing',
+            python_callable=callables['apply_surge_pricing'],
+            provide_context=True,
+        )
+        t_publish = PythonOperator(
+            task_id='publish_results',
+            python_callable=callables['publish_results'],
+            provide_context=True,
+        )
+
+        t_fetch_interactions >> t_compute_demand
+        t_fetch_price_logs >> t_aggregate_prices
+        [t_compute_demand, t_aggregate_prices] >> t_join_features >> t_surge_pricing >> t_publish
+
+    return dag
+
+
+# instantiate DAGs for Airflow to discover
+dag_airline = create_surge_pricing_dag('airline')
+dag_hotel = create_surge_pricing_dag('hotel')

experiments/procesing/pipelines.py

@@ -2,6 +2,7 @@ from sklearn.pipeline import Pipeline
 import pandas as pd
 from procesing.context import PipelineContext
 from procesing.providers import SupabaseProvider, BackendAPIProvider
+import os
 from procesing.steps import (
     FetchInteractionsStep,
     FetchPriceLogsStep,
@@ -12,11 +13,13 @@ from procesing.steps import (
     ChunkByTimeWindowStep,
     ComputeDemandForChunksStep,
     AggregatePriceLogsStep,
-    # BuildStateSpaceStep,
     FitPricingFunctionStep,
     PredictPricesStep,
     ComputeDemandStep,
-    JoinProductFeaturesStep
+    JoinProductFeaturesStep,
+    ExtractSessionFeaturesStep,
+    JoinLabelsStep,
+    ValidateDataStep,
 )
 from procesing.pricers import SimpleSurgePricer
 
@@ -106,33 +109,64 @@ def full_pipeline(context: PipelineContext,
     return product_features_df, optimal_prices_df
 
 
+def ml_training_pipeline(context: PipelineContext) -> pd.DataFrame:
+    """
+    Build labeled session-level feature matrix for ML model training.
+    Pipeline: fetch -> validate -> extract features -> join labels
+
+    Returns:
+        DataFrame with ~25 features per session + is_agent label
+        Columns: sessionId, experimentId, temporal/behavioral/product/ua features, is_agent
+    """
+    # fetch raw interactions
+    interactions_df = FetchInteractionsStep(context).transform(None)
+
+    # validate data quality (report cached in context)
+    interactions_df = ValidateDataStep(context).transform(interactions_df)
+    if interactions_df.empty:
+        return pd.DataFrame()
+
+    # extract vectorized session features
+    features_df = ExtractSessionFeaturesStep(context).transform(interactions_df)
+    if features_df.empty:
+        return pd.DataFrame()
+
+    # join experiment labels (is_agent = ~xp_human_only)
+    labeled_df = JoinLabelsStep(context).transform(features_df)
+
+    return labeled_df
+
+
 
 
 if __name__ == '__main__':
 
-    class Provider(SupabaseProvider, BackendAPIProvider):
+    class ExperimentsProvider(SupabaseProvider, BackendAPIProvider):
-        def __init__(self, backend_url: str):
-            SupabaseProvider.__init__(self)
-            BackendAPIProvider.__init__(self, backend_url=backend_url)
-
-
-    class HistoricalProvider(SupabaseProvider, BackendAPIProvider):
         def fetch_kafka_topic(self, topic: str) -> pd.DataFrame:
-            path = "/home/velocitatem/Documents/Projects/PHANTOM/experiments/collected_data/858c61ab-0a7f-4595-ae49-33f4365517b9/"
+            base_path = "/home/velocitatem/Documents/Projects/PHANTOM/experiments/collected_data/" # os.path.join(os.path.dirname(__file__), "collected_data")
-            interactions_file = "messages(2).json"
+            if not os.path.isdir(base_path):
-            prices_file = "messages(3).json"
+                return pd.DataFrame()
 
-            data = pd.read_json(path + (interactions_file if topic == "user-interactions" else prices_file))
+            files = {"user-interactions": "int.json", "price-logs": "price.json"}
-            data = [r['payload'] for r in data['value'].to_list()]
+            file_to_read = files.get(topic, files["user-interactions"])
-            data = pd.DataFrame(data)
+            frames = []
-            return data
 
+            for d in os.listdir(base_path):
+                full_path = os.path.join(base_path, d, file_to_read)
+                if not os.path.isfile(full_path):
+                    continue
+                try:
+                    data = pd.read_json(full_path)
+                    payloads = pd.DataFrame([r['payload'] for r in data['value'].to_list()])
+                    frames.append(payloads)
+                except Exception as e:
+                    print(f"Warning: Could not process {full_path}: {e}")
 
-    # example run
+            return pd.concat(frames, ignore_index=True) if frames else pd.DataFrame()
-    context = PipelineContext(
-        provider=HistoricalProvider(),
-        store_mode='hotel',
-    )
 
-    product_features, prices = full_pipeline(context)
+    # demo: run ML training pipeline
-    print(prices.to_string())
+    context = PipelineContext(provider=ExperimentsProvider(), store_mode='hotel')
+    features = ml_training_pipeline(context)
+    print(f"Feature matrix: {features.shape}")
+    print(features.head())
+    print(features.info())

experiments/procesing/providers/supabase.py

@@ -18,10 +18,17 @@ class SupabaseProvider(DataProvider):
         self.supabase: Client = create_client(self.supabase_url, self.supabase_key)
 
     def fetch_products(self, store_mode: str) -> pd.DataFrame:
-        resp = self.supabase.table(f'{store_mode}_products').select(
+        # hotel uses room_type, airline uses flight_type; select all and normalize
-            "id, room_type, date_index, metadata, availability"
+        resp = self.supabase.table(f'{store_mode}_products').select("*").execute()
-        ).execute()
+        if not resp.data:
-        return pd.DataFrame(resp.data) if resp.data else pd.DataFrame()
+            return pd.DataFrame()
+        df = pd.DataFrame(resp.data)
+        # normalize type column: hotel has room_type, airline has flight_type
+        if 'room_type' in df.columns:
+            df['product_type'] = df['room_type']
+        elif 'flight_type' in df.columns:
+            df['product_type'] = df['flight_type']
+        return df
 
     def fetch_experiments(self, experiment_ids: List[str]) -> pd.DataFrame:
         if not experiment_ids:

experiments/procesing/steps/__init__.py

@@ -6,7 +6,11 @@ from procesing.steps.chunk import ChunkByTimeWindowStep
 from procesing.steps.demand import ComputeDemandStep, ComputeDemandForChunksStep
 from procesing.steps.elasticity import AggregatePriceLogsStep
 from procesing.steps.pricing import FitPricingFunctionStep, PredictPricesStep
-from procesing.steps.session import ExtractSessionFeaturesStep, _extract_features_for_session
+from procesing.steps.session import (
+    ExtractSessionFeaturesStep, JoinLabelsStep, ValidateDataStep,
+    TemporalFeatureStep, BehavioralFeatureStep, ProductFeatureStep, UserAgentFeatureStep,
+    _extract_features_for_session
+)
 
 __all__ = [
     'BaseContextStep',
@@ -25,5 +29,11 @@ __all__ = [
     'FitPricingFunctionStep',
     'PredictPricesStep',
     'ExtractSessionFeaturesStep',
+    'JoinLabelsStep',
+    'ValidateDataStep',
+    'TemporalFeatureStep',
+    'BehavioralFeatureStep',
+    'ProductFeatureStep',
+    'UserAgentFeatureStep',
     '_extract_features_for_session',
 ]

experiments/procesing/steps/base.py

@@ -1,6 +1,7 @@
 from abc import ABC, abstractmethod
 from sklearn.base import BaseEstimator, TransformerMixin
 from procesing.context import PipelineContext
+from typing import Any
 
 class BaseContextStep(BaseEstimator, TransformerMixin, ABC):
     """
@@ -16,7 +17,7 @@ class BaseContextStep(BaseEstimator, TransformerMixin, ABC):
         return self
 
     @abstractmethod
-    def transform(self, X):
+    def transform(self, X) -> Any:
         """Transform input using context. Must be implemented by subclass."""
         pass
 

experiments/procesing/steps/elasticity.py

@@ -7,12 +7,12 @@ class AggregatePriceLogsStep(BaseContextStep):
     """
     Aggregate price logs into time windows using VECTORIZED operations.
     Input: price_logs_df
-    Output: list of price chunks with [productId, price]
+    Output: DataFrame with columns [productId, price]
     """
 
     def transform(self, price_logs_df: pd.DataFrame):
         if price_logs_df.empty:
-            return []
+            return pd.DataFrame(columns=['productId', 'price'])
 
         df = price_logs_df.copy()
         ts_col = self.context.config.get('ts_col', 'ts')

experiments/procesing/steps/fetch.py

@@ -2,7 +2,7 @@ import pandas as pd
 from procesing.steps.base import BaseContextStep
 
 class FetchInteractionsStep(BaseContextStep):
-    """Fetch raw interaction data from Kafka topic with optional time filtering"""
+    """Fetch raw interaction data from Kafka topic with optional time and store_mode filtering"""
 
     def __init__(self, context, lookback: str = None):
         super().__init__(context)
@@ -24,6 +24,10 @@ class FetchInteractionsStep(BaseContextStep):
         # drop all where page has /admin/
         df = df[~df['page'].str.contains('/admin/', na=False)]
 
+        # filter by store_mode from context
+        if 'storeMode' in df.columns:
+            df = df[df['storeMode'] == self.context.store_mode]
+
         # Remap dateIndex if present
         if 'metadata_dateIndex' in df.columns:
             df['dateIndex'] = df['metadata_dateIndex'].astype('Int64')
@@ -38,7 +42,7 @@ class FetchInteractionsStep(BaseContextStep):
 
 
 class FetchPriceLogsStep(BaseContextStep):
-    """Fetch price log data from Kafka topic with optional time filtering"""
+    """Fetch price log data from Kafka topic with optional time and store_mode filtering"""
 
     def __init__(self, context, lookback: str = None):
         super().__init__(context)
@@ -50,6 +54,10 @@ class FetchPriceLogsStep(BaseContextStep):
         if df.empty:
             return df
 
+        # filter by store_mode from context
+        if 'storeMode' in df.columns:
+            df = df[df['storeMode'] == self.context.store_mode]
+
         # Apply time filtering if lookback specified
         if self.lookback and 'ts' in df.columns:
             df['ts'] = pd.to_datetime(df['ts'])

experiments/procesing/steps/session.py

@@ -1,159 +1,261 @@
 """
-Session feature extraction for S_t component of state space.
+Session feature extraction for ML training pipeline.
-Computes behavioral signals from interaction data already in pipeline.
 """
 import pandas as pd
 import numpy as np
-from typing import Optional, Dict, Any
+import re
-from collections import Counter
+from typing import Dict, Any
 from procesing.steps.base import BaseContextStep
 
-def _extract_features_for_session(session_df: pd.DataFrame, session_timeout_sec: float = 900) -> Dict[str, Any]:
+EVENT_CATS = {
-    """Compute features for single session.
+    'page_view': ['page_view'],
-
+    'item_view': ['view_item_page', 'learn_more_about_item'],
-    Args:
+    'cart_add': ['add_item_to_cart'],
-        session_df: interaction events for this session
+    'purchase': ['purchase', 'checkout_complete'],
-        session_timeout_sec: max gap between events before resetting duration (default 900s = 15min)
+    'hover': ['hover_over_title', 'hover_over_paragraph', 'hover_over_link', 'hover_over_button'],
-    """
+    # 'filter': ['filter', 'search', 'apply_filter'],
-    features = {}
+}
-
+HEADLESS_RE = re.compile(r'HeadlessChrome|Headless|PhantomJS', re.I)
-    # basic counts
+AUTOMATION_RE = re.compile(r'Selenium|Playwright|Puppeteer|WebDriver|chromedriver|geckodriver', re.I)
-    features['total_interactions'] = len(session_df)
+BROWSER_PATTERNS = [('Chrome', r'Chrome/[\d.]+'), ('Firefox', r'Firefox/[\d.]+'),
-
+                    ('Safari', r'Safari/[\d.]+'), ('Edge', r'Edg/[\d.]+')]
-    event_counts = session_df['eventName'].value_counts().to_dict()
-    features['page_views'] = event_counts.get('page_view', 0) + event_counts.get('view_item_page', 0)
-    features['item_views'] = event_counts.get('view_item_page', 0)
-    features['searches'] = event_counts.get('search', 0)
-    features['cart_adds'] = event_counts.get('add_item_to_cart', 0)
-
-    # hover events
-    hover_events = ['hover_over_title', 'hover_over_paragraph', 'hover_over_link', 'hover_over_button']
-    features['hovers'] = sum(event_counts.get(ev, 0) for ev in hover_events)
-
-    # product-level signals
-    product_ids = session_df['productId'].dropna()
-    features['unique_products_viewed'] = product_ids.nunique()
-
-    if len(product_ids) > 0:
-        product_view_counts = Counter(product_ids)
-        features['product_view_depth'] = max(product_view_counts.values())
-    else:
-        features['product_view_depth'] = 0
-
-    # temporal features with session timeout logic
-    if 'ts' in session_df.columns:
-        timestamps = session_df['ts'].sort_values()
-
-        # compute active duration considering timeout gaps
-        if len(timestamps) > 1:
-            time_diffs = timestamps.diff().dropna().dt.total_seconds()
-            # only count gaps shorter than timeout towards active session duration
-            active_diffs = time_diffs[time_diffs <= session_timeout_sec]
-            features['session_duration_sec'] = active_diffs.sum() if len(active_diffs) > 0 else 0.0
-
-            features['avg_time_between_events'] = time_diffs.mean()
-            features['std_time_between_events'] = time_diffs.std()
-        else:
-            features['session_duration_sec'] = 0.0
-            features['avg_time_between_events'] = 0.0
-            features['std_time_between_events'] = 0.0
-
-        if features['session_duration_sec'] > 0:
-            features['interaction_velocity'] = (features['total_interactions'] / features['session_duration_sec']) * 60
-        else:
-            features['interaction_velocity'] = 0.0
-    else:
-        features['session_duration_sec'] = 0.0
-        features['interaction_velocity'] = 0.0
-        features['avg_time_between_events'] = 0.0
-        features['std_time_between_events'] = 0.0
-
-    # cart/conversion signals
-    features['cart_to_view_ratio'] = features['cart_adds'] / features['item_views'] if features['item_views'] > 0 else 0.0
-
-    return features
 
 
-def _apply_to_slice(df: pd.DataFrame) -> pd.DataFrame:
+def _get_browser(s: str) -> str:
-    """Apply feature extraction to sliding window of interactions."""
+    if pd.isna(s): return 'Unknown'
-    # add columns of all features at each step
+    for name, pat in BROWSER_PATTERNS:
-    new_cols = ["total_interactions", "page_views", "item_views", "searches",
+        if re.search(pat, s): return name
-                "cart_adds", "hovers", "unique_products_viewed", "product_view_depth",
+    return 'Other'
-                "session_duration_sec", "interaction_velocity",
-                "avg_time_between_events", "std_time_between_events",
-                "cart_to_view_ratio"]
-    for col in new_cols: df[col] = np.nan
-    for idx in range(1, len(df) + 1):
-        features = _extract_features_for_session(df.iloc[:idx])
-        # fillna kinda meh
-        features = { k: (v if not pd.isna(v) else 0.0) for k, v in features.items() }
-        for col in new_cols:
-            df.at[df.index[idx - 1], col] = features[col]
-        #print(f"Processed {idx}/{len(df)} events for session {df['sessionId'].iloc[0]}")
-    return df
-
-class BuildStateSpaceStep(BaseContextStep):
-    """
-    Build state space representation S_t from session features.
-
-    Input: session_features DataFrame
-    Output: state_space_df DataFrame with S_t vectors
-    """
-
-    def transform(self, rich_dataset: pd.DataFrame) -> pd.DataFrame:
-        # check if features are present
-        required_cols = ["total_interactions", "page_views", "item_views", "searches",
-                         "cart_adds", "hovers", "unique_products_viewed", "product_view_depth",
-                         "session_duration_sec", "interaction_velocity",
-                         "avg_time_between_events", "std_time_between_events",
-                         "cart_to_view_ratio"]
-        if not all(col in rich_dataset.columns for col in required_cols):
-            raise ValueError("Missing required columns for feature extraction.")
-        if rich_dataset.empty:
-            return pd.DataFrame()
 
 
-        # For simplicity, we return as is
+class TemporalFeatureStep(BaseContextStep):
-        return rich_dataset.copy()
+    """Vectorized time-based features: durations, velocities, gaps."""
+
+    def __init__(self, context, timeout_sec: float = 900, velocity_window: str = '5min'):
+        super().__init__(context)
+        self.timeout_sec = timeout_sec
+        self.velocity_window = velocity_window
+
+    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
+        df = X.copy()
+        if df.empty or 'ts' not in df.columns:
+            return pd.DataFrame(columns=pd.Series(['sessionId']))
+
+        df['ts_dt'] = pd.to_datetime(df['ts'])
+        df = df.sort_values(['sessionId', 'ts_dt'])
+        df['time_diff'] = df.groupby('sessionId')['ts_dt'].diff().dt.total_seconds()
+        df['active_diff'] = df['time_diff'].where(df['time_diff'] <= self.timeout_sec, 0)
+
+        agg = df.groupby('sessionId').agg(
+            session_duration_sec=('active_diff', 'sum'),
+            total_interactions=('sessionId', 'count'),
+            avg_time_between_events=('time_diff', 'mean'),
+            std_time_between_events=('time_diff', 'std'),
+            min_time_between_events=('time_diff', 'min'),
+            session_start_hour=('ts_dt', lambda x: x.min().hour),
+        ).reset_index()
+        agg['std_time_between_events'] = agg['std_time_between_events'].fillna(0)
+        agg['interaction_velocity'] = np.where(
+            agg['session_duration_sec'] > 0,
+            (agg['total_interactions'] / agg['session_duration_sec']) * 60, 0)
+
+        vel = df.set_index('ts_dt').groupby('sessionId').resample(self.velocity_window, include_groups=False).size()
+        max_velocity = vel.groupby('sessionId').max().rename('max_velocity_5min')
+        agg = agg.merge(max_velocity, on='sessionId', how='left')
+        agg['max_velocity_5min'] = agg['max_velocity_5min'].fillna(0)
+        return agg
 
 
+class BehavioralFeatureStep(BaseContextStep):
+    """Vectorized event counts and ratios per session."""
+
+    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
+        df = X.copy()
+        if df.empty or 'eventName' not in df.columns:
+            return pd.DataFrame(columns=pd.Series(['sessionId']))
+
+        for cat, events in EVENT_CATS.items():
+            df[f'is_{cat}'] = df['eventName'].isin(events)
+        df['is_hover'] = df['is_hover'] | df['eventName'].str.startswith('hover_over_')
+
+        agg = df.groupby('sessionId').agg(
+            total_events=('eventName', 'count'), unique_pages=('page', 'nunique'),
+            page_views=('is_page_view', 'sum'), item_views=('is_item_view', 'sum'),
+            cart_adds=('is_cart_add', 'sum'), purchases=('is_purchase', 'sum'),
+            hover_events=('is_hover', 'sum'),
+            # filter_events=('is_filter', 'sum'),
+        ).reset_index()
+        agg['cart_to_view_ratio'] = np.where(agg['item_views'] > 0, agg['cart_adds'] / agg['item_views'], 0)
+        agg['conversion_rate'] = np.where(agg['item_views'] > 0, agg['purchases'] / agg['item_views'], 0)
+        agg['hover_intensity'] = np.where(agg['total_events'] > 0, agg['hover_events'] / agg['total_events'], 0)
+        return agg
+
+
+class ProductFeatureStep(BaseContextStep):
+    """Vectorized product interaction features: diversity, depth, price sensitivity."""
+
+    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
+        df = X.copy()
+        if df.empty:
+            return pd.DataFrame(columns=pd.Series(['sessionId']))
+        price_col = next((c for c in ['metadata_base_price', 'metadata_price', 'base_price'] if c in df.columns), None)
+        df['price_seen'] = pd.to_numeric(df[price_col], errors='coerce') if price_col else np.nan
+
+        prod_df = df[df['productId'].notna()]
+        if prod_df.empty:
+            return pd.DataFrame(columns=pd.Series(['sessionId', 'unique_products_viewed', 'product_view_depth', 'avg_price_seen', 'min_price_seen', 'max_price_seen', 'price_range']))
+
+        agg = prod_df.groupby('sessionId').agg(
+            unique_products_viewed=('productId', 'nunique'),
+            product_view_depth=('productId', lambda x: x.value_counts().iloc[0] if len(x) > 0 else 0),
+            avg_price_seen=('price_seen', 'mean'), min_price_seen=('price_seen', 'min'),
+            max_price_seen=('price_seen', 'max'),
+        ).reset_index()
+        agg['price_range'] = (agg['max_price_seen'] - agg['min_price_seen']).fillna(0)
+        return agg
+
+
+class UserAgentFeatureStep(BaseContextStep):
+    """Parse userAgent into bot-detection signals."""
+
+    def transform(self, X: pd.DataFrame) -> pd.DataFrame|pd.Series:
+        df = X.copy()
+        if df.empty or 'userAgent' not in df.columns:
+            return pd.DataFrame(columns=pd.Series(['sessionId']))
+
+        ua = df.groupby('sessionId')['userAgent'].first().reset_index()
+        ua['is_headless'] = ua['userAgent'].str.contains(HEADLESS_RE, na=False)
+        ua['is_automation'] = ua['userAgent'].str.contains(AUTOMATION_RE, na=False)
+        ua['browser_family'] = ua['userAgent'].apply(_get_browser)
+        return ua[['sessionId', 'is_headless', 'is_automation', 'browser_family']]
 
 
 class ExtractSessionFeaturesStep(BaseContextStep):
     """
-    Extract session-level behavioral features from interaction logs.
+    Vectorized session feature extraction - replaces O(n^2) per-row loop.
-
+    Input: interactions_df
-    Input: interactions_df (user-interactions from earlier pipeline step)
+    Output: session-level feature matrix
-    Output: interactions_df with added session feature columns
     """
 
-    def transform(self, interactions_df: pd.DataFrame) -> pd.DataFrame:
+    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
-        if interactions_df.empty:
+        if X.empty:
             return pd.DataFrame()
+        df = X.copy()
 
-        # ensure timestamp column
+        # run all feature steps and merge on sessionId
-        if 'ts' in interactions_df.columns:
+        temporal = TemporalFeatureStep(self.context).transform(df)
-            interactions_df = interactions_df.copy()
+        behavioral = BehavioralFeatureStep(self.context).transform(df)
-            interactions_df['ts'] = pd.to_datetime(interactions_df['ts'])
+        product = ProductFeatureStep(self.context).transform(df)
+        ua = UserAgentFeatureStep(self.context).transform(df)
 
-        # group by session and compute features
+        result = temporal
-        session_features = []
+        for other in [behavioral, product, ua]:
-        for session_id, session_df in interactions_df.groupby('sessionId'):
+            if not other.empty and 'sessionId' in other.columns:
-            new_slice = _apply_to_slice(session_df.sort_values('ts'))
+                result = result.merge(other, on='sessionId', how='left')
-            session_features.append(new_slice)
 
-        return pd.concat(session_features, ignore_index=True)
+        # carry forward experimentId for label joining
+        if 'experimentId' in df.columns:
+            exp_map = df.groupby('sessionId')['experimentId'].first()
+            result = result.merge(exp_map, on='sessionId', how='left')
+
+        return result
 
 
-
+class JoinLabelsStep(BaseContextStep):
-class FilterSessionInteractionsStep(BaseContextStep):
     """
-    Filter interactions DataFrame to specific session.
+    Join experiment labels to session features.
-
+    Input: (features_df, experiments_df) or features_df (fetches experiments)
-    Input: (interactions_df, session_id)
+    Output: labeled feature matrix with is_agent column
-    Output: interactions_df filtered to session_id
     """
 
-    def transform(self, data: tuple) -> pd.DataFrame:
+    def transform(self, X : tuple) -> pd.DataFrame:
-        interactions_df, session_id = data
+        data = X;
-        return interactions_df[interactions_df['sessionId'] == session_id].copy()
+        if isinstance(data, tuple):
+            features_df, experiments_df = data
+        else:
+            features_df = data
+            if 'experimentId' not in features_df.columns:
+                return features_df
+            exp_ids = features_df['experimentId'].dropna().unique().tolist()
+            experiments_df = self.context.provider.fetch_experiments(exp_ids) if exp_ids else pd.DataFrame()
+
+        if features_df.empty:
+            return features_df
+        if experiments_df.empty:
+            features_df['is_agent'] = np.nan
+            return features_df
+
+        exp = experiments_df.copy()
+        if 'id' in exp.columns:
+            exp = exp.rename(columns={'id': 'experimentId'})
+        if 'xp_human_only' in exp.columns:
+            exp['is_agent'] = ~exp['xp_human_only']
+
+        cols = ['experimentId'] + [c for c in ['is_agent', 'xp_human_only', 'xp_market_mode'] if c in exp.columns]
+        return features_df.merge(exp[cols].drop_duplicates(), on='experimentId', how='left')
+
+
+class ValidateDataStep(BaseContextStep):
+    """
+    Data quality checks before training.
+    Input: df
+    Output: df (unchanged, but logs validation report to context)
+    """
+    REQUIRED = ['sessionId', 'eventName', 'ts']
+
+    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
+        df = X.copy()
+        report = {'status': 'valid', 'rows': len(df), 'sessions': 0}
+        if df.empty:
+            report['status'] = 'empty'
+            self.context.cache('validation_report', report)
+            return df
+
+        missing = [c for c in self.REQUIRED if c not in df.columns]
+        if missing:
+            report['status'] = 'invalid'
+            report['missing_cols'] = missing
+
+        report['sessions'] = df['sessionId'].nunique() if 'sessionId' in df.columns else 0
+        report['null_sessions'] = int(df['sessionId'].isna().sum()) if 'sessionId' in df.columns else 0
+        if 'experimentId' in df.columns:
+            report['null_experiments'] = int(df['experimentId'].isna().sum())
+
+        self.context.cache('validation_report', report)
+        return df
+
+
+# legacy compat - kept for backwards compatibility with existing code
+def _extract_features_for_session(session_df: pd.DataFrame, session_timeout_sec: float = 900) -> Dict[str, Any]:
+    """Single-session feature extraction (legacy interface)."""
+    defaults = {k: 0 for k in ['total_interactions', 'page_views', 'item_views', 'searches',
+                               'cart_adds', 'hovers', 'unique_products_viewed', 'product_view_depth',
+                               'session_duration_sec', 'interaction_velocity',
+                               'avg_time_between_events', 'std_time_between_events', 'cart_to_view_ratio']}
+    if session_df.empty:
+        return defaults
+
+    session_df = session_df.copy()
+    if 'sessionId' not in session_df.columns:
+        session_df['sessionId'] = 'tmp'
+
+    # use a dummy context for the steps
+    class DummyCtx: config = {} # should maybe inherit but whatever
+    ctx = DummyCtx()
+
+    t = TemporalFeatureStep(ctx, timeout_sec=session_timeout_sec).transform(session_df)
+    b = BehavioralFeatureStep(ctx).transform(session_df)
+    p = ProductFeatureStep(ctx).transform(session_df)
+
+    result = {}
+    for df in [t, b, p]:
+        if not df.empty:
+            for col in df.columns:
+                if col != 'sessionId':
+                    result[col] = df[col].iloc[0] if len(df) > 0 else 0
+
+    remap = {'hover_events': 'hovers', 'filter_events': 'searches', 'unique_pages': 'unique_pages_visited'}
+    for old, new in remap.items():
+        if old in result:
+            result[new] = result.pop(old)
+    return result

experiments/procesing/tests/conftest.py

@@ -144,7 +144,7 @@ def mock_price_logs_raw_kafka():
                     'price': 162.47,
                     'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
                     'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
-                    'storeMode': 'shop',
+                    'storeMode': 'hotel',
                     'ts': '2025-11-25T21:05:57.967Z'
                 }
             }
@@ -157,7 +157,7 @@ def mock_price_logs_raw_kafka():
                     'price': 743.49,
                     'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
                     'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
-                    'storeMode': 'shop',
+                    'storeMode': 'hotel',
                     'ts': '2025-11-25T21:05:57.993Z'
                 }
             }
@@ -170,7 +170,7 @@ def mock_price_logs_raw_kafka():
                     'price': 163.87,
                     'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
                     'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
-                    'storeMode': 'shop',
+                    'storeMode': 'hotel',
                     'ts': '2025-11-25T21:05:58.009Z'
                 }
             }
@@ -183,7 +183,7 @@ def mock_price_logs_raw_kafka():
                     'price': 397.46,
                     'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
                     'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
-                    'storeMode': 'shop',
+                    'storeMode': 'hotel',
                     'ts': '2025-11-25T21:05:58.049Z'
                 }
             }
@@ -196,7 +196,7 @@ def mock_price_logs_raw_kafka():
                     'price': 401.66,
                     'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
                     'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
-                    'storeMode': 'shop',
+                    'storeMode': 'hotel',
                     'ts': '2025-11-25T21:06:08.864Z'
                 }
             }
@@ -222,7 +222,7 @@ def mock_experiments():
         'created_at': pd.to_datetime(['2025-11-25T20:00:00Z', '2025-11-26T10:00:00Z']),
         'subject_name': ['Session A', 'Session B'],
         'xp_human_only': [True, False],
-        'xp_market_mode': ['hotel', 'shop'],
+        'xp_market_mode': ['hotel', 'airline'],
         'xp_task_id': [None, None]
     })
 
@@ -269,3 +269,13 @@ def empty_context(empty_provider):
         store_mode='hotel',
         window_size='30s'
     )
+
+
+@pytest.fixture
+def session_interactions(mock_interactions):
+    """Enriched interaction data for session feature extraction tests"""
+    df = mock_interactions.copy()
+    df['userAgent'] = ['Mozilla/5.0 Chrome/120', 'Mozilla/5.0 Chrome/120',
+                       'HeadlessChrome/120', 'HeadlessChrome/120', 'HeadlessChrome/120']
+    df['metadata_base_price'] = [None, None, 150.0, 150.0, 200.0]
+    return df

web/src/app/api/ingest/route.ts

@@ -7,7 +7,7 @@ export async function POST(req: NextRequest) {
   try {
     const body = await req.json();
 
-    const storeMode = process.env.STORE_MODE || 'hotel';
+    const storeMode = process.env.NEXT_PUBLIC_STORE_MODE || process.env.STORE_MODE || 'hotel';
     const userAgent = req.headers.get('user-agent') || undefined;
 
     const event: EventBase = {

web/src/app/api/pricing/route.ts

@@ -11,7 +11,7 @@ export async function GET(req: NextRequest) {
     const productId = searchParams.get('productId');
     const sessionId = searchParams.get('sessionId');
     const experimentId = searchParams.get('experimentId');
-    const storeMode = process.env.NEXT_PUBLIC_STORE_MODE || 'shop';
+    const storeMode = process.env.NEXT_PUBLIC_STORE_MODE || process.env.STORE_MODE || 'hotel';
 
     if (!productId) {
         return NextResponse.json(

web/src/components/feats/airline/AirlineHero.tsx

@@ -2,10 +2,20 @@
 
 import { useState, FormEvent } from 'react';
 import { useRouter } from 'next/navigation';
-import { Button, Label, Input, DateInput, RadioGroup, Dropdown, DropdownCounter } from '@/components/ui';
+import { Button, Label, DateInput, Dropdown, DropdownCounter, SelectDropdown, SelectOption } from '@/components/ui';
 import { dateToDaysFromToday } from '@/lib/airline-utils';
 
-type TripType = 'roundtrip' | 'oneway' | 'multicity';
+const CITIES: SelectOption[] = [
+  { value: 'JFK', label: 'New York (JFK)', sublabel: 'John F. Kennedy International' },
+  { value: 'LAX', label: 'Los Angeles (LAX)', sublabel: 'Los Angeles International' },
+  { value: 'ORD', label: 'Chicago (ORD)', sublabel: "O'Hare International" },
+  { value: 'MIA', label: 'Miami (MIA)', sublabel: 'Miami International' },
+  { value: 'SFO', label: 'San Francisco (SFO)', sublabel: 'San Francisco International' },
+  { value: 'SEA', label: 'Seattle (SEA)', sublabel: 'Seattle-Tacoma International' },
+  { value: 'ATL', label: 'Atlanta (ATL)', sublabel: 'Hartsfield-Jackson International' },
+  { value: 'DFW', label: 'Dallas (DFW)', sublabel: 'Dallas/Fort Worth International' },
+];
+
 
 const PlaneIcon = () => (
   <svg className="w-5 h-5" fill="none" stroke="currentColor" viewBox="0 0 24 24">
@@ -22,11 +32,9 @@ const LocationIcon = () => (
 
 export default function AirlineHero() {
   const router = useRouter();
-  const [tripType, setTripType] = useState<TripType>('roundtrip');
   const [origin, setOrigin] = useState('');
   const [destination, setDestination] = useState('');
   const [departDate, setDepartDate] = useState('');
-  const [returnDate, setReturnDate] = useState('');
   const [passengers, setPassengers] = useState({ adults: 1, children: 0, infants: 0 });
 
   const handleSearch = (e: FormEvent) => {
@@ -40,8 +48,6 @@ export default function AirlineHero() {
 
     if (origin) params.set('origin', origin);
     if (destination) params.set('destination', destination);
-    if (tripType !== 'roundtrip') params.set('tripType', tripType);
-    if (returnDate && tripType === 'roundtrip') params.set('returnDate', returnDate);
 
     params.set('adults', passengers.adults.toString());
     params.set('children', passengers.children.toString());
@@ -66,28 +72,15 @@ export default function AirlineHero() {
 
         <div className="search-form">
           <form onSubmit={handleSearch}>
-            <div className="mb-6">
+            <div className="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-3 gap-4">
-              <RadioGroup
-                name="tripType"
-                value={tripType}
-                onChange={setTripType}
-                options={[
-                  { value: 'roundtrip', label: 'Round-trip' },
-                  { value: 'oneway', label: 'One-way' },
-                  { value: 'multicity', label: 'Multi-city' },
-                ]}
-              />
-            </div>
-
-            <div className="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-4 gap-4">
               <div>
                 <Label htmlFor="origin">From</Label>
-                <Input
+                <SelectDropdown
-                  type="text"
                   id="origin"
                   value={origin}
-                  onChange={(e) => setOrigin(e.target.value)}
+                  onChange={setOrigin}
-                  placeholder="Airport or city"
+                  options={CITIES}
+                  placeholder="Select origin"
                   icon={<PlaneIcon />}
                   required
                 />
@@ -95,12 +88,12 @@ export default function AirlineHero() {
 
               <div>
                 <Label htmlFor="destination">To</Label>
-                <Input
+                <SelectDropdown
-                  type="text"
                   id="destination"
                   value={destination}
-                  onChange={(e) => setDestination(e.target.value)}
+                  onChange={setDestination}
-                  placeholder="Airport or city"
+                  options={CITIES}
+                  placeholder="Select destination"
                   icon={<LocationIcon />}
                   required
                 />
@@ -115,20 +108,6 @@ export default function AirlineHero() {
                   required
                 />
               </div>
-
-              <div>
-                <Label htmlFor="returnDate">Return</Label>
-                {tripType === 'roundtrip' ? (
-                  <DateInput
-                    id="returnDate"
-                    value={returnDate}
-                    onChange={(e) => setReturnDate(e.target.value)}
-                    required
-                  />
-                ) : (
-                  <DateInput id="returnDate" disabled />
-                )}
-              </div>
             </div>
 
             <div className="grid grid-cols-4 sm:grid-cols-3 lg:grid-cols-4 gap-4 mt-4">

web/src/components/ui/SelectDropdown.tsx

@@ -0,0 +1,119 @@
+'use client';
+
+import { useState, useRef, useEffect, ReactNode } from 'react';
+
+export interface SelectOption {
+  value: string;
+  label: string;
+  sublabel?: string;
+}
+
+interface SelectDropdownProps {
+  value: string;
+  onChange: (value: string) => void;
+  options: SelectOption[];
+  placeholder?: string;
+  icon?: ReactNode;
+  required?: boolean;
+  id?: string;
+}
+
+export default function SelectDropdown({
+  value,
+  onChange,
+  options,
+  placeholder = 'Select...',
+  icon,
+  required,
+  id,
+}: SelectDropdownProps) {
+  const [open, setOpen] = useState(false);
+  const [filter, setFilter] = useState('');
+  const ref = useRef<HTMLDivElement>(null);
+  const inputRef = useRef<HTMLInputElement>(null);
+
+  useEffect(() => {
+    const handleClick = (e: MouseEvent) => {
+      if (ref.current && !ref.current.contains(e.target as Node)) {
+        setOpen(false);
+        setFilter('');
+      }
+    };
+    document.addEventListener('mousedown', handleClick);
+    return () => document.removeEventListener('mousedown', handleClick);
+  }, []);
+
+  const selectedOption = options.find((o) => o.value === value);
+  const filtered = options.filter(
+    (o) =>
+      o.label.toLowerCase().includes(filter.toLowerCase()) ||
+      o.value.toLowerCase().includes(filter.toLowerCase()) ||
+      o.sublabel?.toLowerCase().includes(filter.toLowerCase())
+  );
+
+  const handleSelect = (opt: SelectOption) => {
+    onChange(opt.value);
+    setOpen(false);
+    setFilter('');
+  };
+
+  return (
+    <div className="relative" ref={ref}>
+      <div
+        className="input-field flex items-center gap-2 cursor-pointer box-border"
+        onClick={() => {
+          setOpen(true);
+          setTimeout(() => inputRef.current?.focus(), 0);
+        }}
+      >
+        {icon && <span className="text-[var(--text-secondary)]">{icon}</span>}
+        {open ? (
+          <input
+            ref={inputRef}
+            type="text"
+            id={id}
+            value={filter}
+            onChange={(e) => setFilter(e.target.value)}
+            placeholder={placeholder}
+            className="flex-1 bg-transparent outline-none text-sm text-[var(--text-primary)]"
+          />
+        ) : (
+          <span className={`flex-1 text-sm ${value ? 'text-[var(--text-primary)]' : 'text-[var(--text-secondary)]'}`}>
+            {selectedOption ? selectedOption.label : placeholder}
+          </span>
+        )}
+        <svg
+          className={`w-4 h-4 text-[var(--text-secondary)] transition-transform ${open ? 'rotate-180' : ''}`}
+          fill="none"
+          stroke="currentColor"
+          viewBox="0 0 24 24"
+        >
+          <path strokeLinecap="round" strokeLinejoin="round" strokeWidth={2} d="M19 9l-7 7-7-7" />
+        </svg>
+      </div>
+      {open && (
+        <div className="absolute z-20 mt-1 w-full bg-[var(--bg-primary)] border-2 border-[var(--accent-primary)] rounded-md shadow-lg max-h-60 overflow-y-auto">
+          {filtered.length === 0 ? (
+            <div className="px-4 py-3 text-sm text-[var(--text-secondary)]">No results</div>
+          ) : (
+            filtered.map((opt) => (
+              <div
+                key={opt.value}
+                onClick={() => handleSelect(opt)}
+                className={`px-4 py-2 cursor-pointer transition-colors hover:bg-[var(--accent-primary-light)] ${
+                  opt.value === value ? 'bg-[var(--accent-primary-light)]' : ''
+                }`}
+              >
+                <div className="text-sm font-medium text-[var(--text-primary)]">{opt.label}</div>
+                {opt.sublabel && <div className="text-xs text-[var(--text-secondary)]">{opt.sublabel}</div>}
+              </div>
+            ))
+          )}
+        </div>
+      )}
+      {required && !value && (
+        <input type="text" required className="sr-only" tabIndex={-1} value="" onChange={() => {}} />
+      )}
+    </div>
+  );
+}

web/src/components/ui/index.ts

@@ -5,3 +5,5 @@ export { default as DateInput } from './DateInput';
 export { default as RadioGroup } from './RadioGroup';
 export { default as Dropdown, DropdownCounter } from './Dropdown';
 export { default as Navigation } from './Navigation';
+export { default as SelectDropdown } from './SelectDropdown';
+export type { SelectOption } from './SelectDropdown';

web/src/lib/config.ts

@@ -16,7 +16,7 @@ const envSchema = z.object({
 // parse and validate env at module load, fail fast with descriptive errors
 const parseEnv = (): Env => {
     const result = envSchema.safeParse({
-        STORE_MODE: process.env.STORE_MODE,
+        STORE_MODE: process.env.NEXT_PUBLIC_STORE_MODE || process.env.STORE_MODE,
         NEXT_PUBLIC_API_BASE: process.env.NEXT_PUBLIC_API_BASE,
         NEXT_PUBLIC_APP_ENV: process.env.NEXT_PUBLIC_APP_ENV,
     });

web/src/styles/airline.css

@@ -278,6 +278,8 @@
   padding: 12px;
   transition: border-color 0.2s ease;
   width: 100%;
+  min-height: 48px;
+  box-sizing: border-box;
 }
 
 [data-mode="airline"] .input-field:focus {