fix: adding date utils

docker-compose.yml

@@ -112,11 +112,14 @@ services:
     depends_on:
       - postgres
     environment:
-      - AIRFLOW__CORE__EXECUTOR=SequentialExecutor
+      - AIRFLOW__CORE__EXECUTOR=LocalExecutor
       - AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@postgres/airflow
       - AIRFLOW__CORE__FERNET_KEY=${AIRFLOW_FERNET_KEY}
       - AIRFLOW__CORE__LOAD_EXAMPLES=false
       - AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
+      - AIRFLOW__CORE__PARALLELISM=16
+      - AIRFLOW__CORE__DAG_CONCURRENCY=8
+      - AIRFLOW__CORE__MAX_ACTIVE_RUNS_PER_DAG=4
       - _AIRFLOW_DB_MIGRATE=true
       - _AIRFLOW_WWW_USER_CREATE=true
       - _AIRFLOW_WWW_USER_USERNAME=admin
@@ -136,12 +139,17 @@ services:
       - airflow-init
       - redis
     environment:
-      - AIRFLOW__CORE__EXECUTOR=SequentialExecutor
+      - AIRFLOW__CORE__EXECUTOR=LocalExecutor
       - AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@postgres/airflow
       - AIRFLOW__CORE__FERNET_KEY=${AIRFLOW_FERNET_KEY}
       - AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION=true
       - AIRFLOW__CORE__LOAD_EXAMPLES=false
       - AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
+      - AIRFLOW__CORE__PARALLELISM=16
+      - AIRFLOW__CORE__DAG_CONCURRENCY=8
+      - AIRFLOW__CORE__MAX_ACTIVE_RUNS_PER_DAG=4
+      - AIRFLOW__SCHEDULER__MIN_FILE_PROCESS_INTERVAL=30
+      - AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL=60
       - AIRFLOW__WEBSERVER__EXPOSE_CONFIG=true
       - AIRFLOW__WEBSERVER__SECRET_KEY=${AIRFLOW_SECRET_KEY}
       - AIRFLOW__API__AUTH_BACKENDS=airflow.api.auth.backend.basic_auth
@@ -174,12 +182,18 @@ services:
       redis:
         condition: service_started
     environment:
-      - AIRFLOW__CORE__EXECUTOR=SequentialExecutor
+      - AIRFLOW__CORE__EXECUTOR=LocalExecutor
       - AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@postgres/airflow
       - AIRFLOW__CORE__FERNET_KEY=${AIRFLOW_FERNET_KEY}
       - AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION=true
       - AIRFLOW__CORE__LOAD_EXAMPLES=false
       - AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
+      - AIRFLOW__CORE__PARALLELISM=16
+      - AIRFLOW__CORE__DAG_CONCURRENCY=8
+      - AIRFLOW__CORE__MAX_ACTIVE_RUNS_PER_DAG=4
+      - AIRFLOW__SCHEDULER__MIN_FILE_PROCESS_INTERVAL=30
+      - AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL=60
+      - AIRFLOW__SCHEDULER__PARSING_PROCESSES=2
       - AIRFLOW__WEBSERVER__SECRET_KEY=${AIRFLOW_SECRET_KEY}
       - AIRFLOW__API__AUTH_BACKENDS=airflow.api.auth.backend.basic_auth
       - KAFKA_HOST=kafka

experiments/airflow/dags/surge_pricing_factory.py

@@ -1,3 +1,4 @@
+from pandas.core.algorithms import factorize_array
 from airflow import DAG
 from airflow.operators.python import PythonOperator
 from airflow.utils.dates import days_ago
@@ -208,3 +209,12 @@ def create_surge_pricing_dag(store_mode: str) -> DAG:
 # instantiate DAGs for Airflow to discover
 dag_airline = create_surge_pricing_dag('airline')
 dag_hotel = create_surge_pricing_dag('hotel')
+
+# TODO: Refactor this factory from a surge pricing factory to a general pricing factory
+# We will do this by passing a pricing strategy class to the factory, since the generic pipeline is:
+# take all interaction data, group by sessionId and assign a new price vector to each session
+# in the grouping we get a subset of the interactions per sessionId and we can map that to some Features
+# we define a custom _get_features(interactions .) methodin the strategy class
+# we then run only the inference which is the .predict(trajectory) per-session which will give us a new price vector
+# this we then publish for each sessionId group
+# this might include no deleting most of the pricers we have defined and starting with a super simple surge-pricing algorithm that is no-fit only predict. This we can then test end-to-end and observe changes to prices according to a desired strategy - we have to define this one as a very short term strategy because we run sessions that take only a few minutes.

experiments/procesing/pricers/base.py

@@ -7,15 +7,6 @@ import pandas as pd
 class PricingFunction(ABC):
     """
     Abstract base for pricing functions.
-
-    Defines mapping: f(Q_t, P_t, S_t, H_t) -> P_{t+1}
-
-    Where:
-        Q_t ∈ R^n: demand vector at time t
-        P_t ∈ R^n: price vector at time t
-        S_t: session features (behavioral signals, interactions)
-        H_t = {Q_{t-k}, P_{t-k}, S_{t-k}}: historical state trajectory
-
     Objective:
         maximize E[R_T] = E[Σ P_t^T · Q_t]
         subject to:
@@ -28,10 +19,10 @@ class PricingFunction(ABC):
     def fit(self, *kwargs):
         """
         Offline training on historical data.
+        This is where we can think about some maximization of expected revenue
+        over historical trajectories to learn parameters of the pricing function.
+        (This however we cover move in the RL side of things)
 
-        Args:
-            historical_data: DataFrame with elasticity, prices, demand signals
-            **kwargs: additional training parameters
         """
         pass
 
@@ -39,12 +30,18 @@ class PricingFunction(ABC):
     def predict(self, *kwargs) -> np.ndarray:
         """
         Generate optimal prices given current state.
+        This is an abstract method that transitions from τ -> P*
+        which is the mapping from the trajectory to optimal prices under
+        some subset of session grouping (so, per sessionId)
+        """
+        pass
 
-        Args:
+    @abstractmethod
-            state_space: StateSpace object containing Q_t, P_t, S_t, H_t
+    def _get_features(self, *kwargs) -> np.ndarray:
-
+        """
+        Extract features from trajectory for pricing decision.
         Returns:
-            P_{t+1}: price vector in R^n
+            np.ndarray of shape (n_products, n_features)
         """
         pass
 

experiments/procesing/pricers/elasticity.py

@@ -57,3 +57,13 @@ class ElasticityBasedPricer(PricingFunction):
         # enforce bounds
         prices = np.clip(prices, self.price_floor, self.price_ceil)
         return prices
+
+    def _get_features(self, state_space=None) -> np.ndarray:
+        """Extract elasticity, demand, and demand deviation for each product"""
+        if state_space is None or self.elasticity is None:
+            n = len(self.elasticity) if self.elasticity is not None else 0
+            return np.zeros((n, 3))
+
+        demand = np.asarray(state_space.demand)
+        demand_dev = (demand - self.mean_demand) / (self.mean_demand + 1e-6)
+        return np.column_stack([self.elasticity, demand, demand_dev])

experiments/procesing/pricers/session_aware.py

@@ -107,6 +107,36 @@ class SessionAwarePricer(PricingFunction):
 
         return prices
 
+    def _get_features(self, state_space=None) -> np.ndarray:
+        """Extract elasticity, demand, and session features"""
+        if state_space is None or self.elasticity is None:
+            n = len(self.elasticity) if self.elasticity is not None else 0
+            return np.zeros((n, 5))
+
+        demand = np.asarray(state_space.demand)
+        n_products = len(demand)
+
+        # extract session features
+        velocity = 0.0
+        view_depth = 0.0
+        cart_to_view = 0.0
+
+        if not state_space.session_features.empty:
+            sf = state_space.session_features.iloc[0]
+            velocity = sf.get('interaction_velocity', 0.0)
+            view_depth = sf.get('product_view_depth', 0.0)
+            cart_to_view = sf.get('cart_to_view_ratio', 0.0)
+
+        # broadcast session features to all products
+        features = np.column_stack([
+            self.elasticity,
+            demand,
+            np.full(n_products, velocity),
+            np.full(n_products, view_depth),
+            np.full(n_products, cart_to_view)
+        ])
+        return features
+
 
 class ProductSpecificSessionPricer(PricingFunction):
     """
@@ -170,3 +200,12 @@ class ProductSpecificSessionPricer(PricingFunction):
 
         prices = np.clip(base_prices, self.price_floor, self.price_ceil)
         return prices
+
+    def _get_features(self, state_space=None) -> np.ndarray:
+        """Extract elasticity and demand features for product-specific pricing"""
+        if state_space is None or self.elasticity is None:
+            n = len(self.elasticity) if self.elasticity is not None else 0
+            return np.zeros((n, 2))
+
+        demand = np.asarray(state_space.demand)
+        return np.column_stack([self.elasticity, demand])

experiments/procesing/pricers/simple.py

@@ -65,6 +65,11 @@ class StaticPricer(PricingFunction):
             raise ValueError("Must call fit() or provide base_prices in constructor")
         return self.base_prices.copy()
 
+    def _get_features(self, state_space=None) -> np.ndarray:
+        """Static pricer uses no features, returns empty array"""
+        n = len(self.base_prices) if self.base_prices is not None else 0
+        return np.zeros((n, 0))
+
 
 class RandomPricer(PricingFunction):
     """Random pricing within bounds (for baseline comparison)"""
@@ -87,6 +92,11 @@ class RandomPricer(PricingFunction):
             self.n_products = len(state_space.demand)
         return self.rng.uniform(self.price_min, self.price_max, size=self.n_products)
 
+    def _get_features(self, state_space=None) -> np.ndarray:
+        """Random pricer uses no features"""
+        n = self.n_products if self.n_products else 0
+        return np.zeros((n, 0))
+
 
 class SimpleSurgePricer(PricingFunction):
     """
@@ -133,3 +143,16 @@ class SimpleSurgePricer(PricingFunction):
         new_prices[low_mask] *= self.discount_multiplier
 
         return new_prices
+
+    def _get_features(self, state_space=None) -> np.ndarray:
+        """Extract demand and base price features for each product"""
+        if state_space is None:
+            n = len(self.base_prices) if self.base_prices is not None else 0
+            return np.zeros((n, 2))
+
+        demand = np.asarray(state_space.demand) if hasattr(state_space, 'demand') else np.array([0])
+        base = np.asarray(state_space.prices) if hasattr(state_space, 'prices') else self.base_prices
+        if base is None:
+            base = np.ones(len(demand)) * 99.99
+
+        return np.column_stack([demand, base])

web/src/app/cart/page.tsx

@@ -32,7 +32,8 @@ export default function CartPage() {
                     {itemCount > 0 && (
                         <button
                             onClick={clearCart}
-                            className="text-sm text-red-600 hover:underline"
+                            className="text-sm hover:underline"
+                            style={{ color: 'var(--accent-warning)' }}
                         >
                             Clear cart
                         </button>
@@ -42,7 +43,7 @@ export default function CartPage() {
                 {itemCount === 0 ? (
                     <div className="text-center py-12">
                         <p className="text-gray-500 mb-4">Your cart is empty</p>
-                        <a href="/" className="text-blue-600 hover:underline">Browse our selection</a>
+                        <a href="/" className="hover:underline" style={{ color: 'var(--text-accent)' }}>Browse our selection</a>
                     </div>
                 ) : (
                     <>
@@ -54,15 +55,11 @@ export default function CartPage() {
                                 >
                                     <div className="flex-1">
                                         <div className="flex items-center gap-2 mb-1">
-                                            <span className="px-2 py-0.5 text-xs font-medium rounded bg-blue-100 text-blue-800">
-                                                {item.type}
-                                            </span>
                                             <h3 className="font-semibold">{item.name}</h3>
                                         </div>
 
                                         {item.type === 'hotel' && (
                                             <div className="text-sm text-gray-600">
-                                                <p>{String(item.metadata.roomType)}</p>
                                                 <p>{String(item.metadata.checkIn)} - {String(item.metadata.checkOut)}</p>
                                                 <p>{String(item.metadata.nights)} night{Number(item.metadata.nights) > 1 ? 's' : ''}</p>
                                             </div>
@@ -81,7 +78,8 @@ export default function CartPage() {
                                         <p className="text-xl font-bold mb-2">${item.price}</p>
                                         <button
                                             onClick={() => handleRemove(item.id, item.type)}
-                                            className="text-sm text-red-600 hover:underline"
+                                            className="text-sm hover:underline"
+                                            style={{ color: 'var(--accent-warning)' }}
                                         >
                                             Remove
                                         </button>
@@ -100,7 +98,7 @@ export default function CartPage() {
                                     dispatchInteraction('checkout_start', undefined, { total, itemCount });
                                     window.location.href = '/checkout';
                                 }}
-                                className="w-full py-3 bg-blue-600 hover:bg-blue-700 text-white rounded-lg font-medium transition-colors"
+                                className="btn-primary w-full"
                             >
                                 Proceed to Checkout
                             </button>

web/src/app/globals.css

@@ -8,6 +8,9 @@
   --bg-secondary: #f5f5f5;
   --text-primary: #333333;
   --text-secondary: #666666;
+  --accent-primary: #007aff;
+  --accent-primary-hover: #0051d5;
+  --accent-primary-light: #e6f2ff;
   --spacing-sm: 8px;
   --spacing-md: 16px;
   --spacing-lg: 32px;

web/src/app/layout.tsx

@@ -15,8 +15,8 @@ const geistMono = Geist_Mono({
 });
 
 export const metadata: Metadata = {
-  title: "Create Next App",
+  title: "Travel Booking Platform",
-  description: "Generated by create next app",
+  description: "Book flights and hotels with dynamic pricing",
 };
 
 export default function RootLayout({

web/src/components/feats/hotel/HotelCard.tsx

@@ -2,6 +2,7 @@
 
 import type { EventName } from '@/lib/events';
 import type { Hotel } from '@/lib/hotel-utils';
+import { getHotelImageUrl } from '@/lib/hotel-utils';
 import { useHoverTracking } from '@/hooks/useHoverTracking';
 import PriceDisplay from '@/components/ui/PriceDisplay';
 
@@ -47,8 +48,6 @@ export default function HotelCard({ hotel }: { hotel: Hotel }) {
         window.location.href = `/hotel/products/${hotel.id}`;
     };
 
-    const imageUrl = `https://images.unsplash.com/photo-1551882547-ff40c63fe5fa?w=400&h=300&fit=crop`;
-
     return (
         <div
             className="hotel-card cursor-pointer"
@@ -56,7 +55,7 @@ export default function HotelCard({ hotel }: { hotel: Hotel }) {
         >
             <div className="hotel-image relative overflow-hidden">
                 <img
-                    src={imageUrl}
+                    src={getHotelImageUrl(hotel.id, { w: 400, h: 300 })}
                     alt={hotel.name}
                     className="w-full h-full object-cover"
                     onError={(e) => {

web/src/components/feats/hotel/HotelDetails.tsx

@@ -2,6 +2,7 @@
 
 import { useState, useEffect } from 'react';
 import type { Hotel } from '@/lib/hotel-utils';
+import { getHotelImageUrl } from '@/lib/hotel-utils';
 import PriceDisplay from '@/components/ui/PriceDisplay';
 
 interface HotelDetailsProps {
@@ -43,13 +44,11 @@ const PriceTotalDisplay = ({ productId, nights }: { productId: string; nights: n
 };
 
 export default function HotelDetails({ product, onAddToCart, addedToCart }: HotelDetailsProps) {
-  const imageUrl = `https://images.unsplash.com/photo-1566073771259-6a8506099945?w=800&h=600&fit=crop`;
-
   return (
     <div className="w-full flex flex-col lg:flex-row gap-12 py-8">
       <div className="w-full lg:w-1/2 rounded-lg aspect-[4/3] overflow-hidden shrink-0">
         <img
-          src={imageUrl}
+          src={getHotelImageUrl(product.id, { w: 800, h: 600 })}
           alt={product.name}
           className="w-full h-full object-cover"
           onError={(e) => {

web/src/components/ui/Navigation.tsx

@@ -20,7 +20,7 @@ const NavLink = ({ href, children }: { href: string; children: React.ReactNode }
       href={href}
       className={`px-4 py-2 rounded-md transition-colors ${
         isActive
-          ? 'bg-[var(--accent-primary)] font-semibold'
+          ? 'bg-[var(--accent-primary)] text-white font-semibold'
           : 'hover:bg-[var(--accent-primary-light)] text-[var(--text-primary)]'
       }`}
     >

web/src/lib/airline-utils.ts

@@ -31,7 +31,7 @@ export interface Flight {
   availability: number;
 }
 
-const EPOCH = new Date(0);
+import { dateToDaysFromToday, dateToIndex, todayIndex } from './date-utils';
 
 export const transformProduct = (p: AirlineProduct): Flight => {
   const { id, flight_type, date_index, metadata, availability } = p;
@@ -52,24 +52,4 @@ export const transformProduct = (p: AirlineProduct): Flight => {
   };
 };
 
-// convert date string to days from today
+export { dateToDaysFromToday, dateToIndex, todayIndex };
-export const dateToDaysFromToday = (dateStr: string): number => {
-  const target = new Date(dateStr);
-  target.setHours(0, 0, 0, 0);
-  const today = new Date();
-  today.setHours(0, 0, 0, 0);
-  return Math.floor((target.getTime() - today.getTime()) / 86400000);
-};
-
-// convert date string to date_index (days since epoch)
-export const dateToIndex = (dateStr: string): number => {
-  const d = new Date(dateStr);
-  return Math.floor((d.getTime() - EPOCH.getTime()) / 86400000);
-};
-
-// get current date_index
-export const todayIndex = (): number => {
-  const now = new Date();
-  now.setHours(0, 0, 0, 0);
-  return Math.floor((now.getTime() - EPOCH.getTime()) / 86400000);
-};

web/src/lib/date-utils.ts

@@ -0,0 +1,23 @@
+const EPOCH = new Date(0);
+const MS_PER_DAY = 86400000;
+
+export const dateToDaysFromToday = (dateStr: string): number => {
+  const target = new Date(dateStr);
+  target.setHours(0, 0, 0, 0);
+  const today = new Date();
+  today.setHours(0, 0, 0, 0);
+  return Math.floor((target.getTime() - today.getTime()) / MS_PER_DAY);
+};
+
+export const dateToIndex = (dateStr: string): number => {
+  const d = new Date(dateStr);
+  return Math.floor((d.getTime() - EPOCH.getTime()) / MS_PER_DAY);
+};
+
+export const todayIndex = (): number => {
+  const now = new Date();
+  now.setHours(0, 0, 0, 0);
+  return Math.floor((now.getTime() - EPOCH.getTime()) / MS_PER_DAY);
+};
+
+export { EPOCH, MS_PER_DAY };

web/src/lib/hotel-utils.ts

@@ -25,7 +25,7 @@ export interface Hotel {
   nights: number;
 }
 
-const EPOCH = new Date(0);
+import { EPOCH, MS_PER_DAY, dateToDaysFromToday, dateToIndex, todayIndex } from './date-utils';
 
 export const transformProduct = (p: HotelProduct): Hotel => {
   const { id, room_type, date_index, metadata } = p;
@@ -37,14 +37,14 @@ export const transformProduct = (p: HotelProduct): Hotel => {
     // legacy: treat as offset from today
     const today = new Date();
     today.setHours(0, 0, 0, 0);
-    checkIn = new Date(today.getTime() + date_index * 86400000);
+    checkIn = new Date(today.getTime() + date_index * MS_PER_DAY);
   } else {
     // proper: days since epoch
-    checkIn = new Date(EPOCH.getTime() + date_index * 86400000);
+    checkIn = new Date(EPOCH.getTime() + date_index * MS_PER_DAY);
   }
 
   const nights = 1;
-  const checkOut = new Date(checkIn.getTime() + nights * 86400000);
+  const checkOut = new Date(checkIn.getTime() + nights * MS_PER_DAY);
 
   const formatOpts: Intl.DateTimeFormatOptions = {
     month: 'short',
@@ -65,24 +65,34 @@ export const transformProduct = (p: HotelProduct): Hotel => {
   };
 };
 
-// convert date string to days from today
+const hotelImagePool = [
-export const dateToDaysFromToday = (dateStr: string): number => {
+  'photo-1566073771259-6a8506099945',
-  const target = new Date(dateStr);
+  'photo-1551882547-ff40c63fe5fa',
-  target.setHours(0, 0, 0, 0);
+  'photo-1590490360182-c33d57733427',
-  const today = new Date();
+  'photo-1582719478250-c89cae4dc85b',
-  today.setHours(0, 0, 0, 0);
+  'photo-1596701062351-8c2c14d1fdd0',
-  return Math.floor((target.getTime() - today.getTime()) / 86400000);
+  'photo-1631049307264-da0ec9d70304',
+  'photo-1578683010236-d716f9a3f461',
+  'photo-1540518614846-7eded433c457',
+  'photo-1505693416388-ac5ce068fe85',
+  'photo-1522771739844-6a9f6d5f14af',
+  'photo-1562438668-bcf0ca6578f0',
+  'photo-1595576508898-0ad5c879a061',
+];
+
+const hashString = (s: string): number => {
+  let h = 0;
+  for (let i = 0; i < s.length; i++) {
+    h = ((h << 5) - h) + s.charCodeAt(i);
+    h = h & h;
+  }
+  return Math.abs(h);
 };
 
-// convert date string to date_index (days since epoch)
+export const getHotelImageUrl = (hotelId: string, size: { w: number; h: number } = { w: 400, h: 300 }): string => {
-export const dateToIndex = (dateStr: string): number => {
+  const idx = hashString(hotelId) % hotelImagePool.length;
-  const d = new Date(dateStr);
+  const photoId = hotelImagePool[idx];
-  return Math.floor((d.getTime() - EPOCH.getTime()) / 86400000);
+  return `https://images.unsplash.com/${photoId}?w=${size.w}&h=${size.h}&fit=crop`;
 };
 
-// get current date_index
+export { dateToDaysFromToday, dateToIndex, todayIndex };
-export const todayIndex = (): number => {
-  const now = new Date();
-  now.setHours(0, 0, 0, 0);
-  return Math.floor((now.getTime() - EPOCH.getTime()) / 86400000);
-};