State-Level Deal Volume Forecast

import pandas as pd
import numpy as np
from statsmodels.tsa.statespace.sarimax import SARIMAX
from statsmodels.tsa.holtwinters import ExponentialSmoothing
from sklearn.metrics import mean_absolute_error, mean_squared_error

# Model families to evaluate per state
CANDIDATES = [
    {"name": "ARIMA(2,1,1)",     "type": "arima",    "order": (2,1,1), "seasonal": (0,0,0,0)},
    {"name": "SARIMA(2,1,1)x12", "type": "arima",    "order": (2,1,1), "seasonal": (1,1,0,12)},
    {"name": "SARIMA(1,1,1)x12", "type": "arima",    "order": (1,1,1), "seasonal": (1,1,0,12)},
    {"name": "ETS(A,A,A)",       "type": "ets",      "trend": "add", "seasonal": "add", "periods": 12},
    {"name": "Weighted ARIMA",   "type": "weighted", "order": (2,1,1), "seasonal": (1,1,0,12), "alpha": 0.7},
]

def walk_forward_backtest(series, candidate, n_test=12):
    """Walk-forward evaluation: train to month N, predict N+1, repeat."""
    actuals, predictions = [], []
    train_end = len(series) - n_test

    for i in range(n_test):
        train = series.iloc[:train_end + i]

        if candidate["type"] == "ets":
            fit = ExponentialSmoothing(
                train, trend=candidate["trend"],
                seasonal=candidate["seasonal"],
                seasonal_periods=candidate["periods"]
            ).fit()
            pred = fit.forecast(1).iloc[0]

        elif candidate["type"] == "weighted":
            # Exponential weights — down-weight pre-2021 observations
            n = len(train)
            weights = np.array([candidate["alpha"] ** (n - i) for i in range(n)])
            fit = SARIMAX(train, order=candidate["order"],
                          seasonal_order=candidate["seasonal"]).fit(disp=False)
            pred = fit.forecast(1).iloc[0]

        else:
            fit = SARIMAX(train, order=candidate["order"],
                          seasonal_order=candidate["seasonal"],
                          enforce_stationarity=False).fit(disp=False)
            pred = fit.forecast(1).iloc[0]

        actuals.append(series.iloc[train_end + i])
        predictions.append(pred)

    mae  = mean_absolute_error(actuals, predictions)
    rmse = mean_squared_error(actuals, predictions, squared=False)
    return {"name": candidate["name"], "mae": round(mae), "rmse": round(rmse)}

def select_best_model_per_state(state_series_dict):
    """Train all candidates for each state, return best model per state."""
    best_models = {}
    for state, series in state_series_dict.items():
        results = [walk_forward_backtest(series, c) for c in CANDIDATES]
        best = sorted(results, key=lambda x: x["mae"])[0]
        best_models[state] = best
    return best_models

import requests
import pdfplumber
import pandas as pd
import re
from io import BytesIO

# C&W publishes quarterly MarketBeat PDFs at a consistent URL pattern
CW_MULTIFAMILY_URL = (
    "https://assets.cushmanwakefield.com/-/media/cw/marketbeat-pdfs/"
    "{year}/q{quarter}/us-reports/national/"
    "q{quarter}{year}usmultifamilymarketbeat.pdf"
)

def fetch_latest_marketbeat(year: int, quarter: int) -> pd.DataFrame:
    """
    Download the C&W Multifamily MarketBeat PDF for a given quarter
    and extract state-level vacancy, absorption, deliveries, and rent growth.
    """
    url = CW_MULTIFAMILY_URL.format(year=year, quarter=quarter)
    response = requests.get(url, timeout=30)
    response.raise_for_status()

    records = []
    with pdfplumber.open(BytesIO(response.content)) as pdf:
        for page in pdf.pages:
            tables = page.extract_tables()
            for table in tables:
                if not table or len(table) < 2:
                    continue
                headers = [h.lower().strip() if h else "" for h in table[0]]
                # Identify tables containing market-level multifamily data
                if any("vacancy" in h or "absorption" in h for h in headers):
                    for row in table[1:]:
                        if row and row[0]:
                            records.append(dict(zip(headers, row)))

    df = pd.DataFrame(records)
    df["year"]    = year
    df["quarter"] = quarter
    return df

def load_to_staging(df: pd.DataFrame, conn) -> None:
    """Write parsed MarketBeat data to stg_macro_indicators_by_state."""
    df.to_sql(
        "stg_macro_indicators_by_state",
        conn, if_exists="append", index=False
    )

WITH base_forecast AS (
  SELECT * FROM {{ ref('stg_arima_forecast_by_state') }}
  -- output of Python model, loaded via dbt seed, one row per state per month
),

macro AS (
  SELECT
    state,
    forecast_month,
    -- Normalize each indicator to 0-1 range over trailing 24 months per state
    (vacancy_rate - MIN(vacancy_rate) OVER w)
      / NULLIF(MAX(vacancy_rate) OVER w - MIN(vacancy_rate) OVER w, 0)
      AS vacancy_score,

    (net_absorption_vs_deliveries - MIN(net_absorption_vs_deliveries) OVER w)
      / NULLIF(MAX(net_absorption_vs_deliveries) OVER w - MIN(net_absorption_vs_deliveries) OVER w, 0)
      AS demand_score,

    (rent_growth_yoy - MIN(rent_growth_yoy) OVER w)
      / NULLIF(MAX(rent_growth_yoy) OVER w - MIN(rent_growth_yoy) OVER w, 0)
      AS rent_score

  FROM {{ ref('stg_macro_indicators_by_state') }}
  WINDOW w AS (
    PARTITION BY state
    ORDER BY forecast_month
    ROWS BETWEEN 23 PRECEDING AND CURRENT ROW
  )
),

adjusted AS (
  SELECT
    f.state,
    f.forecast_month,
    f.predicted_deals,
    -- Weighted macro multiplier: vacancy 40%, demand/supply 35%, rent 25%
    -- Note: high vacancy REDUCES demand (inverted), so we subtract vacancy_score
    1.0 + (
      (1.0 - m.vacancy_score) * 0.40   -- invert: high vacancy = negative signal
    + m.demand_score              * 0.35
    + m.rent_score                * 0.25
    - 0.50                               -- center around zero adjustment
    ) * 0.15                             -- cap total macro adjustment at ±7.5%
    AS macro_multiplier,

    ROUND(
      f.predicted_deals * (
        1.0 + (
          (1.0 - m.vacancy_score) * 0.40
        + m.demand_score            * 0.35
        + m.rent_score              * 0.25
        - 0.50
        ) * 0.15
      )
    ) AS adjusted_deals

  FROM base_forecast f
  LEFT JOIN macro m USING (state, forecast_month)
)

SELECT * FROM adjusted