Blame for engine/damodaran_data.py

"""Aswath Damodaran's published datasets, as data inputs to the valuation.

Source: https://pages.stern.nyu.edu/~adamodar/ (datasets are his; he updates the industry files
each January, the country risk premiums in January and July, and the implied equity risk premium
monthly). Nothing here is fitted to his valuations: these are the same inputs he uses.

* ``build(src_dir)`` parses the downloaded files into ``damodaran_data.json.gz`` (run locally or by
  the refresh chain; needs xlrd/openpyxl). The Space only loads the JSON.
* Loader functions: ``implied_erp``, ``country_premium``, ``company_industry``, ``industry_stats``,
  ``bottom_up_beta``, ``company_erp``.
"""
from __future__ import annotations

import datetime as _dt
import gzip
import json
import os
import re

DATA_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "damodaran_data.json.gz")
BASE_URL = "https://pages.stern.nyu.edu/~adamodar/pc/"
SOURCES = {
    "betas_us": "datasets/betas.xls", "betas_global": "datasets/betaGlobal.xls",
    "capex_us": "datasets/capex.xls", "capex_global": "datasets/capexGlobal.xls",
    "margin_us": "datasets/margin.xls", "margin_global": "datasets/marginGlobal.xls",
    "wacc_us": "datasets/wacc.xls", "wacc_global": "datasets/waccGlobal.xls",
    "ps_us": "datasets/psdata.xls", "ps_global": "datasets/psGlobal.xls",
    "pe_us": "datasets/pedata.xls", "pe_global": "datasets/peGlobal.xls", "companies": "datasets/indname.xls",
    "roe_us": "datasets/roe.xls", "roe_global": "datasets/roeGlobal.xls",
    "country": "datasets/ctrypremJuly26.xlsx", "erp_monthly": "implprem/ERPbymonth.xlsx",
}
# US listings in his company file (exchange prefixes); everything else is a non-US listing
US_EXCHANGES = ("NYSE", "NasdaqGS", "NasdaqGM", "NasdaqCM", "NYSEAM", "ARCA", "BATS", "OTCPK", "OTCQX", "OTCQB", "NYSEArca")
_SUFFIX = re.compile(r"\b(inc|corp|corporation|co|company|ltd|limited|plc|sa|ag|nv|se|holdings?|group|the|llc|lp|"
                     r"incorporated|n\.v|s\.a|a/s|asa|spa|s\.p\.a|kgaa|oyj|ab|bhd|tbk|pcl)\b\.?", re.I)


def _norm_name(name):
    n = re.sub(r"\([^)]*\)", " ", str(name or ""))
    n = _SUFFIX.sub(" ", n.replace("&", " and "))
    return re.sub(r"[^a-z0-9]+", " ", n.lower()).strip()


# ═══════════════════════════════════════════════════════════════════════════════════════
# builder (local / refresh chain)
# ═══════════════════════════════════════════════════════════════════════════════════════
def _num(v):
    try:
        if v is None:
            return None
        if isinstance(v, str):
            pct = "%" in v
            v = v.strip().replace("%", "").replace(",", ".")
            x = float(v) / 100.0 if pct else float(v)
        else:
            x = float(v)
        return None if x != x else x
    except (TypeError, ValueError):
        return None


def _industry_table(path):
    import pandas as pd
    raw = pd.read_excel(path, sheet_name="Industry Averages", header=None)
    h = raw.index[raw.iloc[:, 0].astype(str).str.strip().eq("Industry Name")][0]
    df = raw.iloc[h + 1:].copy()
    df.columns = [str(c).strip() for c in raw.iloc[h]]
    date = raw.iloc[0, 1]
    return df.dropna(subset=["Industry Name"]), (str(date)[:10] if date is not None else None)


def build(src_dir, out=DATA_FILE):
    import pandas as pd
    p = lambda k: os.path.join(src_dir, os.path.basename(SOURCES[k]))
    data = {"meta": {"source": "Aswath Damodaran, https://pages.stern.nyu.edu/~adamodar/",
                     "files": {k: BASE_URL + v for k, v in SOURCES.items()},
                     "built_at": _dt.date.today().isoformat()}}
    # implied ERP by month (his headline figure: trailing-12-month cash yield, column "ERP (T12m)")
    e = pd.read_excel(p("erp_monthly"), sheet_name="Historical ERP")
    rows = []
    for _, r in e.iterrows():
        d, v = r.get("Start of month"), _num(r.get("ERP (T12m)"))
        if d is not None and v is not None and str(d) != "NaT":
            rows.append({"month": str(d)[:10], "erp": round(v, 5), "tbond": _num(r.get("T.Bond Rate"))})
    data["erp"] = {"measure": "implied ERP, trailing 12-month cash yield (ERP (T12m))", "monthly": rows[-36:]}
    # country risk premiums (rating-based default spread × relative equity volatility)
    c = pd.read_excel(p("country"), sheet_name="ERPs by country", header=None)
    h = c.index[c.iloc[:, 0].astype(str).str.strip().eq("Country")][0]
    mature = _num(c.iloc[2, 4])
    countries = {}
    for _, r in c.iloc[h + 1:].iterrows():
        name = str(r.iloc[0]).strip()
        if not name or name == "nan" or _num(r.iloc[5]) is None:
            continue
        countries[name] = {"region": str(r.iloc[1]), "rating": str(r.iloc[2]), "default_spread": _num(r.iloc[3]),
                           "total_erp": _num(r.iloc[4]), "crp": _num(r.iloc[5])}
    data["country"] = {"as_of": str(c.iloc[1, 1])[:10], "mature_erp": mature, "countries": countries}
    # industries: US and global tables
    inds = {}
    market = {}
    for region, keys in (("US", ("betas_us", "capex_us", "margin_us", "ps_us", "wacc_us", "pe_us", "roe_us")),
                         ("Global", ("betas_global", "capex_global", "margin_global", "ps_global", "wacc_global", "pe_global", "roe_global"))):
        b, bd = _industry_table(p(keys[0])); cx, _ = _industry_table(p(keys[1])); m, _ = _industry_table(p(keys[2]))
        ps, _ = _industry_table(p(keys[3])); wc, wd = _industry_table(p(keys[4])); pe, _ = _industry_table(p(keys[5]))
        roe, _ = _industry_table(p(keys[6]))
        cx = cx.set_index("Industry Name"); m = m.set_index("Industry Name"); ps = ps.set_index("Industry Name")
        pe = pe.set_index("Industry Name"); roe = roe.set_index("Industry Name")
        # the mature company: his "Total Market (without financials)" row of the cost-of-capital file, with
        # the Treasury rate and marginal tax rate he estimated it at (so it can be rebuilt at today's rates)
        wraw = pd.read_excel(p(keys[4]), sheet_name="Industry Averages", header=None)
        def lab(s):   # the first number to the right of a label on the input panel
            for _, r in wraw.iterrows():
                if str(r.iloc[0]).strip().startswith(s):
                    return next((_num(v) for v in list(r.iloc[1:8]) if _num(v) is not None), None)
            return None
        tm = wc.set_index("Industry Name").loc["Total Market (without financials)"]
        rf_then = lab("Long Term Treasury bond rate")
        market[region] = {"as_of": wd, "row": "Total Market (without financials)", "beta": _num(tm.get("Beta")),
                          "debt_weight": _num(tm.get("D/(D+E)")), "pre_tax_cost_of_debt": _num(tm.get("Cost of Debt")),
                          "default_spread": (_num(tm.get("Cost of Debt")) - rf_then) if rf_then is not None else None,
                          "marginal_tax": lab("If yes, enter the marginal tax rate"), "rf": rf_then,
                          "erp": lab("Risk Premium to Use for Equity"), "cost_of_capital": _num(tm.get("Cost of Capital")),
                          "n_firms": _num(tm.get("Number of Firms"))}
        tab = {}
        for _, r in b.iterrows():
            n = str(r["Industry Name"]).strip()
            row = {"n_firms": _num(r.get("Number of firms")), "levered_beta": _num(r.get("Beta")), "d_e": _num(r.get("D/E Ratio")),
                   "tax_rate": _num(r.get("Effective Tax rate")), "unlevered_beta": _num(r.get("Unlevered beta")),
                   "unlevered_beta_cash": _num(r.get("Unlevered beta corrected for cash"))}
            if n in cx.index:
                row["sales_to_capital"] = _num(cx.loc[n].get("Sales/ Invested Capital (LTM)"))
            if n in m.index:
                row["operating_margin"] = _num(m.loc[n].get("Pre-tax Unadjusted Operating Margin"))
                row["operating_margin_adj"] = _num(m.loc[n].get("Pre-tax Lease & R&D adj Margin"))
                row["net_margin"] = _num(m.loc[n].get("Net Margin"))
            if n in ps.index:
                row["ev_sales"] = _num(ps.loc[n].get("EV/Sales"))
                row["price_sales"] = _num(ps.loc[n].get("Price/Sales"))
            if n in pe.index:
                # aggregate market cap ÷ trailing net income of the profitable firms: robust to loss-makers, unlike
                # the average of firm P/Es (Software averages 122x with 70% of firms losing money)
                row["pe_aggregate_profitable"] = _num(pe.loc[n].get("Aggregate Mkt Cap/ Trailing Net Income (only money making firms)"))
                row["pe_forward"] = _num(pe.loc[n].get("Forward PE"))
                row["expected_growth_5y"] = _num(pe.loc[n].get("Expected growth - next 5 years"))
            if n in roe.index:
                row["roe"] = _num(roe.loc[n].get("ROE (unadjusted)"))
            tab[n] = row
        inds[region] = {"as_of": bd, "industries": tab}
    data["industries"] = inds
    data["market"] = market
    # companies → industry and country (his classification)
    comp = pd.read_excel(p("companies"), sheet_name="By company name")
    companies, names, sic_counts = {}, {}, {}
    for _, r in comp.iterrows():
        key, ind = str(r.get("Exchange:Ticker") or "").strip(), str(r.get("Industry Group") or "").strip()
        if not key or not ind or ind == "nan":
            continue
        country = str(r.get("Country") or "").strip()
        sic = str(r.get("SIC Code") or "").split(".")[0]
        companies[key] = [ind, country]
        nm = _norm_name(r.get("Company Name"))
        if nm and nm not in names:
            names[nm] = key
        if country == "United States" and sic.isdigit():
            sic_counts.setdefault(sic, {}).setdefault(ind, 0)
            sic_counts[sic][ind] += 1
    data["companies"] = companies
    data["names"] = names
    data["sic_industry"] = {s: max(v.items(), key=lambda kv: kv[1])[0] for s, v in sic_counts.items()}
    with gzip.open(out, "wt", encoding="utf-8") as fh:
        json.dump(data, fh, separators=(",", ":"))
    return {"erp_months": len(rows), "countries": len(countries), "companies": len(companies),
            "industries_us": len(inds["US"]["industries"]), "industries_global": len(inds["Global"]["industries"])}


# ═══════════════════════════════════════════════════════════════════════════════════════
# loader (engine)
# ═══════════════════════════════════════════════════════════════════════════════════════
_CACHE = {"data": None, "loaded": False}
OVERRIDE_FILE = os.path.join(os.getenv("CRUCIBLE_CACHE") or os.path.dirname(os.path.abspath(__file__)), "damodaran_erp_monthly.json")


def _erp_rows():
    rows = list((_data() or {}).get("erp", {}).get("monthly") or [])
    try:
        with open(OVERRIDE_FILE) as fh:
            extra = json.load(fh).get("monthly") or []
        by = {r["month"]: r for r in rows}
        by.update({r["month"]: r for r in extra})
        rows = [by[k] for k in sorted(by)]
    except (OSError, ValueError):
        pass
    return rows


def refresh_erp_if_stale(max_age_days=35, timeout=30):
    """His implied ERP is published monthly. When the newest month we hold is older than
    ``max_age_days``, fetch his monthly file and cache the parsed rows (the bundle stays as is).
    Returns a short status string; a failed fetch leaves the bundled months in use."""
    rows = _erp_rows()
    if rows:
        age = (_dt.date.today() - _dt.date.fromisoformat(rows[-1]["month"][:10])).days
        if age <= max_age_days:
            return f"current ({rows[-1]['month']})"
    try:
        import io
        import pandas as pd
        import requests
        r = requests.get(BASE_URL + SOURCES["erp_monthly"], timeout=timeout)
        r.raise_for_status()
        e = pd.read_excel(io.BytesIO(r.content), sheet_name="Historical ERP")
        new = []
        for _, x in e.iterrows():
            d, v = x.get("Start of month"), _num(x.get("ERP (T12m)"))
            if d is not None and v is not None and str(d) != "NaT":
                new.append({"month": str(d)[:10], "erp": round(v, 5), "tbond": _num(x.get("T.Bond Rate"))})
        with open(OVERRIDE_FILE, "w") as fh:
            json.dump({"fetched": _dt.date.today().isoformat(), "monthly": new[-36:]}, fh)
        return f"refreshed to {new[-1]['month']}" if new else "fetched, no rows"
    except Exception as ex:  # noqa: BLE001
        return f"refresh failed ({str(ex)[:80]}); using {rows[-1]['month'] if rows else 'none'}"


def _data():
    if not _CACHE["loaded"]:
        _CACHE["loaded"] = True
        try:
            with gzip.open(DATA_FILE, "rt", encoding="utf-8") as fh:
                _CACHE["data"] = json.load(fh)
        except (OSError, ValueError):
            _CACHE["data"] = None
    return _CACHE["data"]


def available():
    return _data() is not None


def implied_erp(on=None):
    """{value, month, measure} — his implied ERP for the latest month at or before ``on``."""
    d = _data()
    if not d:
        return None
    rows = _erp_rows()
    if on:
        rows = [r for r in rows if r["month"] <= str(on)[:10]] or rows
    r = rows[-1]
    return {"value": r["erp"], "month": r["month"], "measure": d["erp"]["measure"], "source": d["meta"]["files"]["erp_monthly"]}


_COUNTRY_ALIASES = {"USA": "United States", "US": "United States", "UK": "United Kingdom", "Korea": "Korea",
                    "South Korea": "Korea", "Republic of Korea": "Korea", "Hong Kong SAR": "Hong Kong", "Netherlands (the)": "Netherlands"}


def country_premium(country):
    """{crp, rating, total_erp, as_of} for a country of operations, or None when not in his table."""
    d = _data()
    if not d or not country:
        return None
    name = _COUNTRY_ALIASES.get(str(country).strip(), str(country).strip())
    row = d["country"]["countries"].get(name)
    if row is None:
        low = name.lower()
        row = next((v for k, v in d["country"]["countries"].items() if k.lower() == low or k.lower().startswith(low)), None)
    return dict(row, country=name, as_of=d["country"]["as_of"], mature_erp=d["country"]["mature_erp"]) if row else None


def company_industry(ticker, name=None, sic=None):
    """(industry, country, how) from his company list: the US listing ticker first, then the company
    name (foreign issuers are listed under their home ticker), then the SIC code's modal industry."""
    d = _data()
    if not d:
        return None, None, None
    t = str(ticker or "").upper()
    for ex in US_EXCHANGES:
        hit = d["companies"].get(f"{ex}:{t}")
        if hit:
            return hit[0], hit[1], f"ticker {ex}:{t}"
    nm = _norm_name(name)
    if nm and nm in d["names"]:
        k = d["names"][nm]
        ind, ctry = d["companies"][k]
        return ind, ctry, f"company name → {k}"
    s = str(sic or "").split(".")[0]
    if s and s in d["sic_industry"]:
        return d["sic_industry"][s], None, f"SIC {s} (modal industry)"
    return None, None, None


def industry_stats(industry, region="US"):
    d = _data()
    if not d or not industry:
        return None
    tab = d["industries"].get(region) or d["industries"]["US"]
    row = tab["industries"].get(industry)
    return dict(row, industry=industry, region=region, as_of=tab["as_of"]) if row else None


def bottom_up_beta(stats, debt, market_cap, tax=0.25):
    """His bottom-up beta: the industry's unlevered beta corrected for cash, re-levered at the
    company's market debt-to-equity: βL = βu × (1 + (1 − t) × D/E)."""
    if not stats:
        return None
    bu = stats.get("unlevered_beta_cash") or stats.get("unlevered_beta")
    if bu is None:
        return None
    de = (float(debt) / float(market_cap)) if (debt and market_cap and market_cap > 0) else 0.0
    return float(bu) * (1.0 + (1.0 - tax) * min(max(de, 0.0), 5.0))


def company_erp(country, on=None):
    """The ERP for a company operating in ``country``: his implied (mature-market) ERP plus the
    country risk premium of the country of operations (zero for the US and Aaa countries)."""
    base = implied_erp(on)
    if base is None:
        return None
    cp = country_premium(country) if country else None
    us = str(country or "United States").strip() in ("United States", "USA", "US", "")
    crp = 0.0 if (us or cp is None) else float(cp.get("crp") or 0.0)
    return {"erp": base["value"] + crp, "implied_erp": base["value"], "crp": crp, "country": (cp or {}).get("country") or country,
            "rating": (cp or {}).get("rating"), "erp_month": base["month"], "crp_as_of": (cp or {}).get("as_of"),
            "basis": ("implied ERP + country risk premium of the country of operations" if crp else "implied ERP (mature market)")}


def industry_multiples(industry, region="US"):
    """Sourced industry pricing anchors: {pe: {value, basis, source, date, universe}, ps: {...}} or {}."""
    st = industry_stats(industry, region) or industry_stats(industry, "US") or {}
    d = _data() or {}
    out = {}
    if st.get("pe_aggregate_profitable"):
        out["pe"] = {"value": st["pe_aggregate_profitable"], "basis": "TRAILING_GAAP (aggregate market cap ÷ trailing net income, "
                     "profitable firms)", "source": "Damodaran pedata/peGlobal", "date": st.get("as_of"),
                     "universe": f"{st.get('region')} {industry} ({int(st.get('n_firms') or 0)} firms)"}
    if st.get("price_sales"):
        out["ps"] = {"value": st["price_sales"], "basis": "TRAILING (aggregate market cap ÷ trailing revenue)",
                     "source": "Damodaran psdata/psGlobal", "date": st.get("as_of"),
                     "universe": f"{st.get('region')} {industry} ({int(st.get('n_firms') or 0)} firms)"}
    return out


# industries too broad or economically mixed for their beta to stand as company-specific terminal evidence
BROAD_INDUSTRIES = ("Diversified", "Business & Consumer Services", "Conglomerates", "Retail (General)",
                    "Financial Svcs. (Non-bank & Insurance)", "Information Services", "Total Market")
_FINANCIAL_WORDS = ("bank", "financial svcs", "insurance", "brokerage", "investments & asset", "reit")
BETA_SANE = (0.5, 2.0)


def beta_quality(inputs, is_financial=False):
    """{status: HIGH|LOW, reasons[], beta, beta_source, beta_industry}: may this beta stand as evidence of the
    company's own risk in stable growth? Checks the industry mapping, the business-model fit of the industry,
    the business mix and the beta's statistical sanity. Peer betas are not in the data, so peer consistency is
    recorded as not assessed."""
    di = inputs or {}
    ind = str(di.get("industry") or "")
    b = di.get("beta")
    reasons = []
    if not ind:
        reasons.append("no industry classification")
    if di.get("industry_match") and not str(di.get("industry_match")).startswith("ticker"):
        reasons.append(f"industry matched by {di.get('industry_match')}, not by ticker")
    if ind in BROAD_INDUSTRIES:
        reasons.append(f"industry '{ind}' is broad or economically mixed")
    if not is_financial and any(w in ind.lower() for w in _FINANCIAL_WORDS):
        reasons.append(f"a financial-services industry beta ('{ind}') for a non-financial business: "
                       "financial firms' debt is operating, so their unlevered betas are not comparable")
    if b is None or not (BETA_SANE[0] <= float(b) <= BETA_SANE[1]):
        reasons.append(f"beta {b if b is None else round(float(b), 2)} outside the plausible range {BETA_SANE[0]}–{BETA_SANE[1]}")
    mix = di.get("business_mix") or []
    if len({r.get("industry") for r in mix}) > 2:
        reasons.append("business mix spans more than two industries")
    # every industry the business-mix beta used must come through a compatible mapping (not only the company's own)
    mix_bad = business_mix_problems(mix, ind or None, is_financial)
    reasons += [f"business-mix segment mapping incompatible: {x}" for x in mix_bad]
    return {"status": "LOW" if reasons else "HIGH", "reasons": reasons, "beta": b,
            "business_mix_beta_valid": (not mix_bad) if mix else None,
            "beta_source": di.get("beta_basis"), "beta_industry": ind or None,
            "peer_consistency": "not assessed (no peer betas in the data)"}


def mature_cost_of_capital(rf, erp, region="US", tax=None):
    """The cost of capital of a mature company, his stable-growth convention ("roughly that of an average
    company"): the beta, debt ratio and default spread of his Total Market (without financials) row,
    rebuilt at today's riskfree rate and the company's ERP. ``None`` when the dataset lacks the row."""
    d = _data()
    mk = ((d or {}).get("market") or {}).get(region) or ((d or {}).get("market") or {}).get("US")
    if not mk or rf is None or erp is None or mk.get("beta") is None or mk.get("debt_weight") is None:
        return None
    t = mk.get("marginal_tax") if tax is None else tax
    t = 0.25 if t is None else float(t)
    ke = float(rf) + float(mk["beta"]) * float(erp)
    kd = (float(rf) + float(mk.get("default_spread") or 0.0)) * (1.0 - t)
    w = float(mk["debt_weight"])
    return {"value": (1.0 - w) * ke + w * kd, "cost_of_equity": ke, "after_tax_cost_of_debt": kd, "beta": mk["beta"],
            "debt_weight": w, "default_spread": mk.get("default_spread"), "tax": t, "region": region, "as_of": mk.get("as_of"),
            "basis": f"mature company: {mk['row']} ({region}, {mk.get('as_of')}) at today's riskfree rate and ERP"}


def download(dest):
    """Fetch every source file into ``dest`` (the refresh chain calls this, then ``build``)."""
    import requests
    os.makedirs(dest, exist_ok=True)
    got = {}
    for k, path in SOURCES.items():
        r = requests.get(BASE_URL + path, timeout=120)
        r.raise_for_status()
        with open(os.path.join(dest, os.path.basename(path)), "wb") as fh:
            fh.write(r.content)
        got[k] = {"bytes": len(r.content), "last_modified": r.headers.get("last-modified")}
    return got


if __name__ == "__main__":
    import sys
    dest = sys.argv[1] if len(sys.argv) > 1 else "damodaran_src"
    print(json.dumps(download(dest), indent=1))
    print(json.dumps(build(dest), indent=1))


# ── segments → his industries (deterministic; first compatible match wins; unmatched → the company's industry) ──
def _kw(*tokens, stems=()):
    """One keyword pattern. Every TOKEN matches only as a whole word (bounded on both sides: "card" never matches
    "cardiovascular", "gas" never "gastroenterology", "power" never "powertrain"). A STEM is declared explicitly and
    matches a word that begins with it ("pharma" → pharmaceutical)."""
    parts = list(tokens) + [s + r"\w*" for s in stems]
    return r"\b(?:" + "|".join(parts) + r")\b"


SEGMENT_INDUSTRY = [
    (_kw("launch(?:es)?", "space", "rockets?", "aerospace", "defen[cs]e", "aircraft", "aviation", "avionics"), "Aerospace/Defense"),
    (_kw("connectivity", "broadband", "starlink", "wireless", "internet services?", "satellite internet", stems=("telecom",)),
     "Telecom. Services"),
    (_kw("cloud", "aws", "azure", "software", "saas", "subscription software", "productivity", "business processes"),
     "Software (System & Application)"),
    (_kw("ai", "artificial intelligence", "machine learning", "llm", "generative"), "Software (System & Application)"),
    (_kw(stems=("advertis",)), "Advertising"),
    (_kw("search", "social", "family of apps", "online services"), "Software (Internet)"),
    # "retail" alone is not a business (retail electricity, retail banking): only store/merchandise retail maps here
    (_kw("online stores?", "physical stores?", "retail stores?", "e-?commerce", "third[- ]party sellers?", "merchandise",
         "general merchandise"), "Retail (General)"),
    (_kw("semiconductors?", "chips?", "gpus?", "compute and networking", "data cent(?:er|re)s?"), "Semiconductor"),
    (_kw("automotive", "vehicles?", "auto"), "Auto & Truck"),
    (_kw("energy generation", "energy storage", "solar", "renewables?"), "Green & Renewable Energy"),
    (_kw("drugs?", "oncology", "vaccines?", stems=("pharma", "biotech", "therapeut")), "Drugs (Pharmaceutical)"),
    (_kw("medical", "medtech", "diagnostics?", "devices? health", "healthcare products"), "Healthcare Products"),
    (_kw("gaming", "video games?", "xbox"), "Software (Entertainment)"),
    (_kw("media", "entertainment", "studios?", "streaming", "content", "parks"), "Entertainment"),
    (_kw("cable"), "Cable TV"),
    (_kw("beverages?"), "Beverage (Soft)"),
    (_kw("foods?", "snacks?"), "Food Processing"),
    (_kw("hotels?", "casinos?", "resorts?", "cruises?"), "Hotel/Gaming"),
    (_kw("restaurants?"), "Restaurant/Dining"),
    (_kw("oil", "gas", "upstream", "downstream", "refining", "chemicals? and energy"), "Oil/Gas (Integrated)"),
    (_kw("chemicals?"), "Chemical (Diversified)"),
    (_kw("mining", "metals", "copper", "gold"), "Metals & Mining"),
    (_kw("machinery", "industrial automation", "automation", "motion"), "Machinery"),
    (_kw("buildings?", "hvac", "construction"), "Building Materials"),
    (_kw("power", "electric(?:ity)?", "utility", "utilities", "retail energy", "wholesale energy", "energy charge"), "Power"),
    (_kw("iphone", "mac", "ipad", "wearables", "computers?", "pcs?", "hardware", "personal computing"), "Computers/Peripherals"),
    (_kw("insurance"), "Insurance (General)"),
    (_kw("payments?", "cards?"), "Financial Svcs. (Non-bank & Insurance)"),
]
_IGNORE_SEGMENTS = re.compile(r"\b(corporate|eliminations?|unallocated|reconcil)", re.I)
# a segment named only by a place is a region of the company's own business, not a different business
_REGION_NAME = re.compile(r"^(?:(?:north|south|east|west|central|northeast|northwest|southeast|southwest|midwest|mid-?atlantic|gulf|"
                          r"pacific|atlantic|americas?|latin|europe(?:an)?|emea|asia|apac|greater|china|japan|india|canada|mexico|"
                          r"texas|california|florida|new york|international|domestic|united|states|kingdom|u\.?s\.?|uk|rest|of|the|"
                          r"world|and|other|region|regions|markets?|operations|segment|&|\s|-)+)$", re.I)


def keyword_matches(text):
    """Every (industry, pattern, matched words) the text matches, in table order."""
    low = str(text or "").lower()
    out = []
    for rx, ind in SEGMENT_INDUSTRY:
        m = re.search(rx, low)
        if m:
            out.append((ind, rx, m.group(0)))
    return out


def segment_industry(name, product_lines=(), fallback=None):
    """(industry, how) for one reportable segment: its name first, then its product-line names. A segment
    named only by a region is the company's own business in that region."""
    if _REGION_NAME.match(str(name or "").strip()):
        return fallback, "regional segment: company industry"
    for text, how in ((name, "segment name"), (" ".join(product_lines or ()), "product lines")):
        hits = keyword_matches(text)
        if hits:
            return hits[0][0], f"{how} matches {hits[0][0]}"
    return fallback, "no keyword match: company industry"


# ── parent-sector compatibility: what a segment's industry may be, given the company it belongs to ──
PASS, REJECTED, UNAVAILABLE_MAPPING = "PASS", "REJECTED", "UNAVAILABLE"
# explicit evidence that a segment is itself a financial business: one of these stands as a whole word in its own text
_FINANCIAL_EVIDENCE = _kw("financial", "finance", "financing", "credit", "bank", "banking", "lending", "leasing", "insurance",
                          "reinsurance", "brokerage", "payments?", "cards?", "asset management", "wealth management")


def is_financial_industry(industry):
    return any(w in str(industry or "").lower() for w in _FINANCIAL_WORDS)


def industry_compatible(parent_industry, parent_is_financial, industry, segment_text):
    """(ok, reason): may a segment of this company carry this industry? A clearly non-financial parent's segment may
    carry a financial industry only on explicit evidence in the segment's own text (a financial word standing as a whole
    word): financial firms' debt is operating, so their EV/Sales and unlevered betas do not describe an operating
    business. No other parent/segment pair is restricted in this pass."""
    if not industry:
        return False, "no industry"
    parent_fin = bool(parent_is_financial) or is_financial_industry(parent_industry)
    if is_financial_industry(industry) and not parent_fin:
        if not re.search(_FINANCIAL_EVIDENCE, str(segment_text or "").lower()):
            return False, (f"financial industry '{industry}' for a segment of a non-financial company "
                           f"({parent_industry or 'unclassified'}) with no financial business named in its own text")
    return True, ""


def map_segment(segment_id, name, product_lines=(), parent_industry=None, parent_is_financial=False, use_product_lines=True):
    """SegmentIndustryMapping for one reportable segment. Fallback hierarchy after a rejection:
    1. the next keyword match in the segment's own text that is compatible with the parent (a validated segment industry);
    2. the company-level industry (compatible with itself);
    3. a broader compatible sector: not available (his datasets carry no sector level above the industry);
    4. mapping unavailable (no selected industry: the segment may feed neither value weights nor betas)."""
    text = str(name or "")
    m = {"segment_id": segment_id, "input_text": text, "selected_industry": None, "mapping_method": None,
         "matched_pattern": None, "matched_text": None, "parent_industry": parent_industry,
         "compatibility_status": UNAVAILABLE_MAPPING, "fallback_used": None, "rejected": [],
         "eligible_for_value_weight": False, "eligible_for_beta": False}

    def select(industry, method, pattern=None, matched=None, fallback=None):
        m.update(selected_industry=industry, mapping_method=method, matched_pattern=pattern, matched_text=matched,
                 compatibility_status=PASS, fallback_used=fallback, eligible_for_value_weight=True, eligible_for_beta=True)
        return m

    if _REGION_NAME.match(text.strip()):
        if parent_industry:
            return select(parent_industry, "regional segment: company industry", fallback="COMPANY_INDUSTRY")
        m["mapping_method"] = "regional segment: no company industry"
        return m
    sources = [(text, "segment name")]
    if use_product_lines and product_lines:
        sources.append((" ".join(str(x) for x in product_lines if x), "product lines"))
    for src, how in sources:
        for ind, rx, matched in keyword_matches(src):
            ok, why = industry_compatible(parent_industry, parent_is_financial, ind, src)
            if ok:
                return select(ind, f"{how} matches {ind}", rx, matched)
            m["rejected"].append({"industry": ind, "source": how, "matched_pattern": rx, "matched_text": matched,
                                  "compatibility_status": REJECTED, "reason": why})
    if parent_industry:
        return select(parent_industry, ("company industry (its keyword match was rejected as incompatible)" if m["rejected"]
                                        else "no keyword match: company industry"), fallback="COMPANY_INDUSTRY")
    m["mapping_method"] = "no compatible industry and no company industry"
    m["fallback_used"] = "UNAVAILABLE"
    return m


class IncompatibleMapping(Exception):
    """A segment's industry row was requested through a mapping that did not pass compatibility."""


def mapping_problem(mapping, parent_is_financial=False):
    """Why a SegmentIndustryMapping may not supply an industry row (None = usable). Its own status must be PASS and
    the pair it records is re-checked, so a record that merely says PASS cannot carry an incompatible industry."""
    mp = mapping or {}
    if mp.get("compatibility_status") != PASS or not mp.get("selected_industry"):
        return f"mapping status {mp.get('compatibility_status')}"
    ok, why = industry_compatible(mp.get("parent_industry"), parent_is_financial, mp.get("selected_industry"),
                                  mp.get("input_text"))
    return None if ok else why


def industry_row(mapping, region, use, parent_is_financial=False):
    """The industry statistics a mapping may supply for ``use`` ("value_weight": EV/Sales; "beta": unlevered beta). The
    ONE accessor the business mix reads, so a bad mapping cannot be amplified through weight and beta together."""
    why = mapping_problem(mapping, parent_is_financial)
    flag = {"value_weight": "eligible_for_value_weight", "beta": "eligible_for_beta"}[use]
    if why is None and not (mapping or {}).get(flag):
        why = f"mapping not eligible for {use}"
    if why:
        raise IncompatibleMapping(f"{(mapping or {}).get('segment_id') or (mapping or {}).get('input_text')}: {why}")
    ind = mapping["selected_industry"]
    return industry_stats(ind, region) or industry_stats(ind, "US") or {}


def business_mix(segments, region, company_industry, debt, market_cap, tax=0.25, parent_is_financial=False):
    """His business-mix beta: each business line's industry unlevered beta (cash-corrected), weighted
    by the ESTIMATED VALUE of the business (its revenue × its industry's EV/Sales, his weighting), and
    re-levered at the company's market D/E. Revenue weights are used only when an EV/Sales is missing,
    and the basis says so. Also returns the industry sales-to-capital of the mix (capital = Σ revenue /
    sales-to-capital) and each segment's industry benchmarks. None unless 2+ business lines.

    Every row reads its industry through a SegmentIndustryMapping that passed compatibility (industry_row): the
    same industry supplies the EV/Sales weight AND the beta, so an incompatible mapping is refused for both and the
    whole mix is withheld (the single-industry beta then applies)."""
    rows = []
    for g in segments or []:
        if _IGNORE_SEGMENTS.search(str(g.get("name") or "")) or not (g.get("revenue") or 0) > 0:
            continue
        if g.get("allocations") is not None:
            # canonical segments (segments.allocate): one row per industry allocation, fractions sum to 1
            allocs = [(a["industry"], a["basis"], float(a["fraction"]),
                       a.get("mapping") or {"segment_id": g.get("segment_id"), "input_text": g.get("name"),
                                            "selected_industry": a["industry"], "parent_industry": company_industry,
                                            "compatibility_status": PASS, "eligible_for_value_weight": True,
                                            "eligible_for_beta": True}) for a in g["allocations"]]
        else:
            if g.get("kind") == "geographic":
                continue
            mp = map_segment(g.get("segment_id"), g.get("name"), [p.get("name") for p in g.get("product_lines") or []],
                             company_industry, parent_is_financial)
            if not mp["selected_industry"]:
                return None
            allocs = [(mp["selected_industry"], mp["mapping_method"], 1.0, mp)]
        for ind, how, fr, mp in allocs:
            try:
                st_w = industry_row(mp, region, "value_weight", parent_is_financial)
                st_b = industry_row(mp, region, "beta", parent_is_financial)
            except IncompatibleMapping:
                return None              # never a precision blend over a mapping that failed compatibility
            rows.append({"name": g.get("name") + (f" ({ind})" if len(allocs) > 1 else ""), "segment_id": g.get("segment_id"),
                         "fraction": fr, "revenue": g["revenue"] * fr, "industry": ind, "match": how,
                         "unlevered_beta": st_b.get("unlevered_beta_cash") or st_b.get("unlevered_beta"),
                         "industry_sales_to_capital": st_b.get("sales_to_capital"), "industry_operating_margin": st_b.get("operating_margin"),
                         "industry_ev_sales": st_w.get("ev_sales"),
                         "mapping": {k: mp.get(k) for k in ("segment_id", "input_text", "selected_industry", "mapping_method",
                                                            "matched_pattern", "matched_text", "parent_industry",
                                                            "compatibility_status", "fallback_used",
                                                            "eligible_for_value_weight", "eligible_for_beta")}})
    if len(rows) < 2 or any(r["unlevered_beta"] is None for r in rows) or len({r["industry"] for r in rows}) < 2:
        return None          # one business, or several lines of the same business: the single-industry beta applies
    tot = sum(r["revenue"] for r in rows)
    by_value = all((r["industry_ev_sales"] or 0) > 0 for r in rows)
    for r in rows:
        r["revenue_weight"] = r["revenue"] / tot
        r["estimated_value"] = (r["revenue"] * r["industry_ev_sales"]) if by_value else None
    vtot = sum(r["estimated_value"] for r in rows) if by_value else None
    for r in rows:
        r["weight"] = (r["estimated_value"] / vtot) if by_value else r["revenue_weight"]
    bu = sum(r["weight"] * r["unlevered_beta"] for r in rows)
    s2c = [r for r in rows if r["industry_sales_to_capital"]]
    s2c_mix = (sum(r["revenue"] for r in s2c) / sum(r["revenue"] / r["industry_sales_to_capital"] for r in s2c)) if s2c else None
    de = (float(debt) / float(market_cap)) if (debt and market_cap and market_cap > 0) else 0.0
    return {"segments": rows, "unlevered_beta": bu, "beta": bu * (1.0 + (1.0 - tax) * min(max(de, 0.0), 5.0)),
            "sales_to_capital": s2c_mix, "weighting": "value" if by_value else "revenue",
            "weights": ("estimated value: segment revenue × its industry EV/Sales (his datasets)" if by_value else
                        "share of revenue (an industry EV/Sales is missing, so value weights could not be formed)")}


def business_mix_problems(mix_rows, company_industry, parent_is_financial=False):
    """[problem] for the rows of a business-mix beta: every industry it used must come through a compatible mapping
    (BUSINESS_MIX_BETA_VALID). A row with no mapping record is re-checked from its own industry and name."""
    bad = []
    for r in mix_rows or []:
        mp = r.get("mapping") or {"segment_id": r.get("segment_id"), "input_text": str(r.get("name") or "").split(" (")[0],
                                  "selected_industry": r.get("industry"), "parent_industry": company_industry,
                                  "compatibility_status": PASS}
        mp = dict(mp, selected_industry=r.get("industry"))          # the industry the row actually used
        why = mapping_problem(mp, parent_is_financial)
        if why:
            bad.append(f"{r.get('name')}: {why}")
    return bad


# Different economics (his reason for valuing a company business by business): reported operating
# margins far apart, or businesses whose industries carry very different risk.
SEGMENT_MARGIN_SPREAD = 0.20
SEGMENT_BETA_SPREAD = 0.25


def segment_build_evidence(filed_segments, mix_rows=None):
    """{required, kind, reason, low, high, low_label, high_label}. A segment-by-segment story is required
    when the company files two or more business lines whose reported operating margins span at least
    SEGMENT_MARGIN_SPREAD ("margin"), or whose industries' unlevered betas span at least
    SEGMENT_BETA_SPREAD ("risk"). The numbers are kept as fields so a report can cite them as claims."""
    segs = [g for g in filed_segments or [] if g.get("kind", "business_line") == "business_line"]
    if len(segs) < 2:
        return {"required": False, "kind": None, "reason": "fewer than two business lines filed"}
    m = [(g.get("name"), g["margin"]) for g in segs if g.get("margin") is not None]
    if len(m) >= 2:
        lo, hi = min(m, key=lambda x: x[1]), max(m, key=lambda x: x[1])
        if hi[1] - lo[1] >= SEGMENT_MARGIN_SPREAD:
            return {"required": True, "kind": "margin", "low": lo[1], "high": hi[1], "low_label": lo[0], "high_label": hi[0],
                    "reason": f"reported operating margins span {lo[1]*100:.0f}% ({lo[0]}) to {hi[1]*100:.0f}% ({hi[0]})"}
    # risk is compared between REPORTABLE SEGMENTS: a segment allocated to several industries (segments.allocate)
    # carries its fraction-weighted beta, so a product-line split inside one segment is not a second business
    per = {}
    for r in mix_rows or []:
        if r.get("unlevered_beta") is None:
            continue
        k = r.get("segment_id") or r.get("name") or id(r)
        a = per.setdefault(k, {"label": r.get("industry"), "w": 0.0, "bw": 0.0, "n": 0})
        fr = float(r.get("fraction") or 1.0)
        a["w"] += fr; a["bw"] += fr * r["unlevered_beta"]; a["n"] += 1
        if a["n"] > 1:
            a["label"] = str(r.get("name") or k).split(" (")[0] + " (mixed)"
    b = [(a["label"], a["bw"] / a["w"]) for a in per.values() if a["w"] > 0]
    if len(b) >= 2:
        lo, hi = min(b, key=lambda x: x[1]), max(b, key=lambda x: x[1])
        if hi[1] - lo[1] >= SEGMENT_BETA_SPREAD:
            return {"required": True, "kind": "risk", "low": lo[1], "high": hi[1], "low_label": lo[0], "high_label": hi[0],
                    "reason": f"industry risk differs: {lo[0]} unlevered beta {lo[1]:.2f} vs {hi[0]} {hi[1]:.2f}"}
    return {"required": False, "kind": None, "reason": "business lines have similar margins and industry risk"}


def segment_build_required(filed_segments, mix_rows=None):
    """(required, reason) — see ``segment_build_evidence``."""
    e = segment_build_evidence(filed_segments, mix_rows)
    return e["required"], e["reason"]