""" One-shot script that builds `demo.sqlite` with synthetic market-performance data. What it does: 1. Loads the anonymized parameter CSVs from `parameters/` into the SQLite parameter tables (cluster_mapping, market_summary_mapping, threshold_matrix, country_region_mapping, company_products). 2. Generates fully fictional metrics rows for every (Region × Cluster × Market × Calculation_Type × Period) combination declared in the parameters. 3. Writes everything to `demo.sqlite` (replacing existing contents). All numbers are produced with `numpy.random` using a fixed seed so the demo is reproducible. No real-world data is used. """ from __future__ import annotations import sys from pathlib import Path import numpy as np import pandas as pd HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) from db import ( # noqa: E402 (sys.path edit above) create_all_tables, insert_metrics, replace_metrics, replace_param_table, ) PARAMS_DIR = HERE / "parameters" PERIODS: list[str] = ["24Q4", "25Q1", "25Q2", "25Q3", "25Q4"] CALCULATION_TYPES: list[str] = ["MAT_YoY", "MAT_QoQ", "QTR_YoY", "QTR_QoQ"] # Market → TA Market (therapeutic area) mapping. MARKET_TO_TA: dict[str, str] = { "FSH/hMG Market": "FE", "Growth Hormone Market": "FE", "Hypothyroid Market": "FE", "Hyperthyroid Market": "FE", "MS High Efficacy Market": "N&I", "Injectable Platform Market": "N&I", "Anti-EGFR Market": "ONC", "MET Inhibitor Market": "ONC", "Beta Blocker Market": "CV", "Antihypertensive Combo Market": "CV", } # Which NovaPharma product (if any) competes in each market, plus fully fictional # competitor products. None of the brand names exist in the real world. MARKET_PRODUCTS: dict[str, list[str]] = { "FSH/hMG Market": ["NOVAFERT-A", "NOVAFERT-B", "OVAFOL", "FOLLINOR", "MENOSTIM"], "Growth Hormone Market": ["NOVATROPIN", "GROWSTAT", "STATUREX", "PEDISOM", "GENOTROPE"], "Hypothyroid Market": ["NOVATHYRO", "THYRESOL", "LEVOSTIM", "THYROCID", "LEVOFORM"], "Hyperthyroid Market": ["NOVAZOL", "METHIMOL", "CARBITHY", "PROPYTHY", "TIAMAZO"], "MS High Efficacy Market": ["NOVALAD", "OCRELIN", "TYSALINE", "KESINAL", "ZEPOSIO"], "Injectable Platform Market": ["NOVASCLER", "INTERON", "BETAGEN", "AVOCOXIN", "PEGSTAT"], "Anti-EGFR Market": ["NOVAGFR", "PANIMAX", "CETUVEXX", "EGFRULIN", "NIMOTUXX"], "MET Inhibitor Market": ["NOVAMETI", "CAPVARIN", "METARIB", "METVARIO", "CRISETIB"], "Beta Blocker Market": ["NOVACOR", "PROPRANOL", "METOPROL", "ATENOROL", "BISOPROL"], "Antihypertensive Combo Market": ["NOVACOMBO", "AMLODIL", "LISINOX", "VALSARTO", "RAMIPROX"], } # How big each region is, roughly (multiplier on absolute value/volume). REGION_SIZE: dict[str, float] = { "NA": 1.0, "EU": 0.95, "APAC": 0.75, "LATAM": 0.45, "MEAR": 0.40, "Global": 2.5, } RNG = np.random.default_rng(seed=2025) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _load_csv(filename: str) -> pd.DataFrame: path = PARAMS_DIR / filename return pd.read_csv(path, keep_default_na=False) def _split_countries(value: str) -> list[str]: return [c.strip() for c in str(value).split(",") if c.strip()] def _allocate_shares(n_products: int) -> np.ndarray: """Random market shares summing to ~100 across competitors (TOTAL row excluded).""" raw = RNG.dirichlet(alpha=np.ones(n_products) * 1.5) * 100.0 return np.round(raw, 2) def _ranks_from_shares(shares: np.ndarray) -> np.ndarray: """Lower rank = larger share.""" order = np.argsort(-shares, kind="stable") ranks = np.empty_like(order) ranks[order] = np.arange(1, len(shares) + 1) return ranks def _round_value(x: float) -> float: """Round monetary value to a nice 2-decimal number.""" return float(np.round(x, 2)) def _signed_growth(low: float, high: float) -> float: """Sample a percentage growth in [low, high] and round to 2 dp.""" return float(np.round(RNG.uniform(low, high), 2)) # --------------------------------------------------------------------------- # Row generators # --------------------------------------------------------------------------- def _make_product_rows( *, period: str, region: str, level: str, country_field: str, cluster_field: str, market: str, calc_type: str, base_value: float, base_volume: float, ) -> list[dict]: products = MARKET_PRODUCTS[market] ta_market = MARKET_TO_TA[market] value_shares = _allocate_shares(len(products)) volume_shares = _allocate_shares(len(products)) value_ranks = _ranks_from_shares(value_shares) volume_ranks = _ranks_from_shares(volume_shares) rows: list[dict] = [] # Per-product rows for i, product in enumerate(products): v_share = float(value_shares[i]) vol_share = float(volume_shares[i]) v_rank = int(value_ranks[i]) vol_rank = int(volume_ranks[i]) v_growth = _signed_growth(-18.0, 28.0) vol_growth = _signed_growth(-15.0, 22.0) asp_growth = _signed_growth(-4.0, 6.0) current_value = _round_value(base_value * v_share / 100.0) current_volume = _round_value(base_volume * vol_share / 100.0) rows.append({ "Period": period, "Region": region, "Level": level, "Product": product, "Country": country_field, "Cluster": cluster_field, "TA Market": ta_market, "Class": market, "Current_Value": current_value, "Current_Value_MS": v_share, "Value_MS_Change": _signed_growth(-3.5, 3.5), "Current_Value_Rank": v_rank, "Value_Rank_Change": int(RNG.integers(-2, 3)), "Value_Growth": v_growth, "Value_Trend_Reversal": float(RNG.choice([0.0, 1.0], p=[0.92, 0.08])), "Current_Volume": current_volume, "Current_Volume_MS": vol_share, "Volume_MS_Change": _signed_growth(-3.5, 3.5), "Current_Volume_Rank": vol_rank, "Volume_Rank_Change": int(RNG.integers(-2, 3)), "Volume_Growth": vol_growth, "Volume_Trend_Reversal": float(RNG.choice([0.0, 1.0], p=[0.92, 0.08])), "ASP_Growth": asp_growth, "Calculation_Type": calc_type, }) # TOTAL row (aggregate of the whole market) — values 100% of base, shares 100% rows.append({ "Period": period, "Region": region, "Level": level, "Product": "TOTAL", "Country": country_field, "Cluster": cluster_field, "TA Market": ta_market, "Class": market, "Current_Value": _round_value(base_value), "Current_Value_MS": 100.0, "Value_MS_Change": 0.0, "Current_Value_Rank": 0, "Value_Rank_Change": 0, "Value_Growth": _signed_growth(-8.0, 14.0), "Value_Trend_Reversal": 0.0, "Current_Volume": _round_value(base_volume), "Current_Volume_MS": 100.0, "Volume_MS_Change": 0.0, "Current_Volume_Rank": 0, "Volume_Rank_Change": 0, "Volume_Growth": _signed_growth(-6.0, 12.0), "Volume_Trend_Reversal": 0.0, "ASP_Growth": _signed_growth(-3.0, 5.0), "Calculation_Type": calc_type, }) return rows def _base_value(region: str, n_countries: int) -> float: """Plausible cluster-level market value in USD millions.""" size = REGION_SIZE.get(region, 0.5) return float(np.round(50.0 * size * (0.6 + 0.4 * n_countries) * RNG.uniform(0.8, 1.4), 2)) def _base_volume(region: str, n_countries: int) -> float: """Plausible cluster-level volume in units (millions).""" size = REGION_SIZE.get(region, 0.5) return float(np.round(2.0 * size * (0.6 + 0.4 * n_countries) * RNG.uniform(0.7, 1.5), 2)) # --------------------------------------------------------------------------- # Main seeding loop # --------------------------------------------------------------------------- def seed_parameters() -> None: """Replace SQLite parameter tables with the contents of the CSV seeds.""" print(" → seeding parameter tables …") replace_param_table("cluster_mapping", _load_csv("cluster_mapping.csv")) replace_param_table("market_summary_mapping", _load_csv("market_summary_mapping.csv")) replace_param_table("threshold_matrix", _load_csv("threshold_matrix.csv")) replace_param_table("country_region_mapping", _load_csv("country_region_mapping.csv")) replace_param_table("company_products", _load_csv("company_products.csv")) def seed_metrics(batch_size: int = 5000) -> int: """Generate synthetic metrics rows from cluster_mapping and stream into SQLite.""" cluster_df = _load_csv("cluster_mapping.csv") buffer: list[dict] = [] total = 0 def flush() -> None: nonlocal total if not buffer: return n = insert_metrics(pd.DataFrame(buffer)) total += n buffer.clear() # Truncate metrics first (replace_metrics deletes then no-ops on empty input) replace_metrics(pd.DataFrame()) for _, row in cluster_df.iterrows(): market = str(row["market"]).strip() cluster = str(row["cluster"]).strip() region = str(row["region"]).strip() if market not in MARKET_PRODUCTS: continue countries = _split_countries(row["country"]) if not countries: continue country_csv = ", ".join(countries) for calc_type in CALCULATION_TYPES: for period in PERIODS: base_value_cluster = _base_value(region, len(countries)) base_volume_cluster = _base_volume(region, len(countries)) # Cluster-level rows buffer.extend(_make_product_rows( period=period, region=region, level="Cluster", country_field=country_csv, cluster_field=cluster, market=market, calc_type=calc_type, base_value=base_value_cluster, base_volume=base_volume_cluster, )) # Country-level rows, splitting the cluster total across countries country_weights = RNG.dirichlet(alpha=np.ones(len(countries)) * 2.0) for c_idx, country in enumerate(countries): weight = float(country_weights[c_idx]) buffer.extend(_make_product_rows( period=period, region=region, level="Country", country_field=country, cluster_field=country, market=market, calc_type=calc_type, base_value=base_value_cluster * weight, base_volume=base_volume_cluster * weight, )) if len(buffer) >= batch_size: flush() flush() return total def main() -> None: print("Creating SQLite schema …") create_all_tables() seed_parameters() print("Generating synthetic metrics rows (this takes a few seconds) …") total_rows = seed_metrics() print(f"Done. Inserted {total_rows:,} rows into sentinel_metrics.") print(f"Database: {Path('demo.sqlite').resolve()}") if __name__ == "__main__": main()