{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Taipei Border Premium: Local Conditional Premium\n",
    "\n",
    "This is a **result reproduction notebook** for the public Taipei border premium artifact. It is not a raw-data full reproduction notebook: it starts from the published analysis artifacts and model-ready input, then re-runs the main local conditional premium estimate and checks the published expected results.\n",
    "\n",
    "Public artifacts used here:\n",
    "\n",
    "- `summary.json`: QA, balance, robustness, and placebo summaries.\n",
    "- `expected_results.json`: expected numeric checks for this result artifact.\n",
    "- `model_input.csv.gz`: public model-ready regression input."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import annotations\n",
    "\n",
    "import json\n",
    "from io import BytesIO\n",
    "from typing import Iterable\n",
    "from urllib.request import Request, urlopen\n",
    "\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import statsmodels.formula.api as smf\n",
    "from IPython.display import display\n",
    "\n",
    "SUMMARY_URL = \"https://pyecontech.com/data/taipei-border-premium/summary.json\"\n",
    "EXPECTED_RESULTS_URL = \"https://pyecontech.com/data/taipei-border-premium/expected_results.json\"\n",
    "MODEL_INPUT_URL = \"https://pyecontech.com/data/taipei-border-premium/model_input.csv.gz\"\n",
    "HTTP_HEADERS = {\n",
    "    \"User-Agent\": \"Mozilla/5.0 (compatible; PyInvest-Colab-Repro/1.0)\"\n",
    "}\n",
    "\n",
    "EXPECTED_PREMIUM_PCT = 25.9144\n",
    "EXPECTED_NOBS = 1322\n",
    "PREMIUM_TOLERANCE = 1e-3"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Load Public Artifacts\n",
    "\n",
    "The notebook intentionally reads only public URLs. It does not depend on private repositories, local files, or raw-data build steps."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def open_public_url(url: str):\n",
    "    \"\"\"Open a public URL with a browser-like User-Agent.\"\"\"\n",
    "    return urlopen(Request(url, headers=HTTP_HEADERS))\n",
    "\n",
    "\n",
    "def read_json_url(url: str) -> dict:\n",
    "    \"\"\"Read a public JSON URL into a dictionary.\"\"\"\n",
    "    with open_public_url(url) as response:\n",
    "        return json.load(response)\n",
    "\n",
    "\n",
    "def read_csv_url(url: str) -> pd.DataFrame:\n",
    "    \"\"\"Read the public gzipped CSV model frame.\"\"\"\n",
    "    with open_public_url(url) as response:\n",
    "        return pd.read_csv(BytesIO(response.read()), compression=\"gzip\")\n",
    "\n",
    "\n",
    "summary = read_json_url(SUMMARY_URL)\n",
    "expected_results = read_json_url(EXPECTED_RESULTS_URL)\n",
    "model_input = read_csv_url(MODEL_INPUT_URL)\n",
    "\n",
    "print(f\"Loaded {len(model_input):,} rows and {len(model_input.columns):,} columns.\")\n",
    "display(model_input.head())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def nested_get(mapping: dict, keys: Iterable[str], default: object = None) -> object:\n",
    "    \"\"\"Read a nested dictionary value using a sequence of keys.\"\"\"\n",
    "    value: object = mapping\n",
    "    for key in keys:\n",
    "        if not isinstance(value, dict) or key not in value:\n",
    "            return default\n",
    "        value = value[key]\n",
    "    return value\n",
    "\n",
    "\n",
    "def log_coef_to_percent(coef: float) -> float:\n",
    "    \"\"\"Convert a log-point coefficient into a percent premium.\"\"\"\n",
    "    return float(np.expm1(coef) * 100.0)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Main Estimate\n",
    "\n",
    "The formula below matches the published `control_parking` specification:\n",
    "\n",
    "`log_unit_price_ping ~ is_taipei + distance + Taipei-side distance slope + property controls + month fixed effects + segment fixed effects`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "required_columns = [\n",
    "    \"log_unit_price_ping\",\n",
    "    \"is_taipei\",\n",
    "    \"dist_100m\",\n",
    "    \"area_building_ping\",\n",
    "    \"building_age\",\n",
    "    \"floor\",\n",
    "    \"total_floor\",\n",
    "    \"building_type\",\n",
    "    \"main_use\",\n",
    "    \"material\",\n",
    "    \"year_month\",\n",
    "    \"boundary_segment_id\",\n",
    "    \"has_parking_int\",\n",
    "]\n",
    "missing = sorted(set(required_columns) - set(model_input.columns))\n",
    "if missing:\n",
    "    raise KeyError(f\"model_input.csv.gz is missing required columns: {missing}\")\n",
    "\n",
    "formula = expected_results[\"target_model\"][\"formula\"]\n",
    "print(formula)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "regression_data = model_input.dropna(subset=required_columns).copy()\n",
    "model = smf.ols(formula=formula, data=regression_data).fit(cov_type=\"HC1\")\n",
    "\n",
    "premium_coef = float(model.params[\"is_taipei\"])\n",
    "premium_pct = log_coef_to_percent(premium_coef)\n",
    "\n",
    "result_table = pd.DataFrame(\n",
    "    {\n",
    "        \"metric\": [\"premium_pct\", \"coef\", \"std_error\", \"nobs\"],\n",
    "        \"value\": [\n",
    "            premium_pct,\n",
    "            premium_coef,\n",
    "            float(model.bse[\"is_taipei\"]),\n",
    "            int(model.nobs),\n",
    "        ],\n",
    "    }\n",
    ")\n",
    "display(result_table)\n",
    "print(model.summary().tables[1])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Expected Result Assertions\n",
    "\n",
    "The checks below assert the published headline result, QA gate status, and estimation sample size."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "expected_premium = float(\n",
    "    expected_results.get(\n",
    "        \"premium_pct\",\n",
    "        nested_get(expected_results, [\"main_estimate\", \"premium_pct\"], EXPECTED_PREMIUM_PCT),\n",
    "    )\n",
    ")\n",
    "expected_nobs = int(\n",
    "    expected_results.get(\n",
    "        \"n\",\n",
    "        expected_results.get(\n",
    "            \"N\",\n",
    "            nested_get(expected_results, [\"main_estimate\", \"n\"], EXPECTED_NOBS),\n",
    "        ),\n",
    "    )\n",
    ")\n",
    "\n",
    "failed_gates = expected_results.get(\"failed_gates_expected\", [])\n",
    "qa_gates = summary.get(\"qa_gates\", {})\n",
    "qa_pass = bool(qa_gates) and all(bool(value) for value in qa_gates.values())\n",
    "\n",
    "assert np.isclose(premium_pct, expected_premium, atol=PREMIUM_TOLERANCE), (\n",
    "    f\"premium_pct mismatch: got {premium_pct:.6f}, \"\n",
    "    f\"expected {expected_premium:.6f}\"\n",
    ")\n",
    "assert int(model.nobs) == expected_nobs, (\n",
    "    f\"N mismatch: got {int(model.nobs)}, expected {expected_nobs}\"\n",
    ")\n",
    "assert failed_gates == [], f\"Expected no failed QA gates, got {failed_gates}\"\n",
    "assert bool(qa_pass), \"Expected QA pass status\"\n",
    "\n",
    "print(\"All expected result checks passed.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Balance, Robustness, and Placebo Summaries\n",
    "\n",
    "These sections are read from `summary.json`. They are displayed rather than rebuilt from raw inputs because this notebook reproduces the result artifact, not the full raw-data pipeline."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sections = {\n",
    "    \"BALANCE\": summary.get(\"balance_table\", []),\n",
    "    \"ROBUSTNESS\": summary.get(\"robustness_control_parking\", []),\n",
    "    \"PLACEBO\": summary.get(\"same_side_placebo_checks\", []),\n",
    "}\n",
    "\n",
    "for section_name, section in sections.items():\n",
    "    print(f\"\\n{section_name}\")\n",
    "    display(pd.DataFrame(section))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}