"""
Data Prefetch Module for US Stock Analysis

Pre-fetches US stock data using yfinance (via USDataClient) to inject into agent
instructions, eliminating the need for yahoo_finance MCP server tool calls.

This reduces token usage by avoiding MCP tool call round-trips for predictable,
parameterized data fetches (OHLCV, holder info, market indices).
"""

import logging
from pathlib import Path
import importlib.util

import numpy as np
import pandas as pd

logger = logging.getLogger(__name__)


def _df_to_markdown(df: pd.DataFrame, title: str = "") -> str:
    """Convert DataFrame to markdown table string (no tabulate dependency).

    Args:
        df: DataFrame to convert
        title: Optional title to prepend

    Returns:
        Markdown table string
    """
    if df is None or df.empty:
        return f"### {title}\n\n_No data available_\n" if title else "_No data available_\n"

    result = ""
    if title:
        result += f"### {title}\n\n"

    # Build markdown table without requiring tabulate
    index_name = str(df.index.name) if df.index.name else "Index"
    headers = [index_name] + [str(c) for c in df.columns]
    result += "| " + " | ".join(headers) + " |\n"
    result += "| " + " | ".join(["---"] * len(headers)) + " |\n"
    for idx, row in df.iterrows():
        cells = [str(idx)] + [str(v) for v in row]
        result += "| " + " | ".join(cells) + " |\n"
    return result


def _get_us_data_client():
    """Get USDataClient instance, importing from the local module.

    Returns:
        USDataClient instance
    """
    # Import USDataClient from the same directory
    _current_dir = Path(__file__).parent
    _client_path = _current_dir / "us_data_client.py"
    spec = importlib.util.spec_from_file_location("us_data_client_local", _client_path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module.USDataClient()


def prefetch_us_stock_ohlcv(ticker: str, period: str = "1y") -> str:
    """Prefetch US stock OHLCV data using yfinance.

    Args:
        ticker: Stock ticker symbol (e.g., "AAPL") or index symbol (e.g., "^GSPC")
        period: Data period (default: "1y")

    Returns:
        Markdown formatted OHLCV data string, or empty string on error
    """
    try:
        client = _get_us_data_client()
        df = client.get_ohlcv(ticker, period=period, interval="1d")

        if df is None or df.empty:
            logger.warning(f"No OHLCV data for {ticker}")
            return ""

        # Capitalize column names for readability
        df.columns = [col.title().replace("_", " ") for col in df.columns]
        df.index.name = "Date"

        return _df_to_markdown(df, f"OHLCV: {ticker} ({period})")
    except Exception as e:
        logger.error(f"Error prefetching OHLCV for {ticker}: {e}")
        return ""


def prefetch_us_holder_info(ticker: str) -> str:
    """Prefetch US institutional holder data using yfinance.

    Args:
        ticker: Stock ticker symbol

    Returns:
        Markdown formatted holder data string (major + institutional + mutualfund), or empty string on error
    """
    try:
        client = _get_us_data_client()
        holders = client.get_institutional_holders(ticker)

        if not holders:
            logger.warning(f"No holder data for {ticker}")
            return ""

        result = ""

        # Major holders
        major = holders.get("major_holders")
        if major is not None and not major.empty:
            result += _df_to_markdown(major, f"Major Holders: {ticker}")
            result += "\n"

        # Institutional holders
        institutional = holders.get("institutional_holders")
        if institutional is not None and not institutional.empty:
            result += _df_to_markdown(institutional, f"Top Institutional Holders: {ticker}")
            result += "\n"

        # Mutual fund holders
        mutualfund = holders.get("mutualfund_holders")
        if mutualfund is not None and not mutualfund.empty:
            result += _df_to_markdown(mutualfund, f"Top Mutual Fund Holders: {ticker}")
            result += "\n"

        return result if result else ""
    except Exception as e:
        logger.error(f"Error prefetching holder info for {ticker}: {e}")
        return ""


def prefetch_us_market_indices(reference_date: str = None) -> dict:
    """Prefetch US market index data.

    Args:
        reference_date: Reference date (YYYYMMDD) - used for logging only

    Returns:
        Dictionary with index data as markdown strings:
        - "sp500": S&P 500 data
        - "nasdaq": NASDAQ Composite data
        - "dow": Dow Jones data
        - "russell": Russell 2000 data
        - "vix": VIX data
    """
    indices = {
        "sp500": ("^GSPC", "1y"),
        "nasdaq": ("^IXIC", "1y"),
        "dow": ("^DJI", "1y"),
        "russell": ("^RUT", "1y"),
        "vix": ("^VIX", "3mo"),
    }

    result = {}
    for key, (symbol, period) in indices.items():
        data = prefetch_us_stock_ohlcv(symbol, period=period)
        if data:
            result[key] = data

    if result:
        logger.info(f"Prefetched US market indices: {list(result.keys())}")

    return result


def prefetch_stock_info(ticker: str) -> str:
    """Prefetch company info and key statistics via yfinance.

    Replaces yahoo_finance MCP get_stock_info call and
    firecrawl key-statistics/financials page scrapes.

    Args:
        ticker: Stock ticker symbol

    Returns:
        Markdown formatted company info string, or empty string on error
    """
    try:
        client = _get_us_data_client()
        info = client.get_company_info(ticker)

        if not info or not info.get("name"):
            logger.warning(f"No company info for {ticker}")
            return ""

        def _fmt(val, fmt_type="default"):
            if val is None or val == 0:
                return "N/A"
            if fmt_type == "currency":
                if abs(val) >= 1e12:
                    return f"${val/1e12:.2f}T"
                elif abs(val) >= 1e9:
                    return f"${val/1e9:.2f}B"
                elif abs(val) >= 1e6:
                    return f"${val/1e6:.2f}M"
                return f"${val:,.2f}"
            elif fmt_type == "percent":
                return f"{val*100:.2f}%" if abs(val) < 1 else f"{val:.2f}%"
            elif fmt_type == "ratio":
                return f"{val:.2f}"
            elif fmt_type == "number":
                if abs(val) >= 1e9:
                    return f"{val/1e9:.2f}B"
                elif abs(val) >= 1e6:
                    return f"{val/1e6:.2f}M"
                return f"{val:,.0f}"
            return str(val)

        result = f"### Company Info: {info.get('name', ticker)} ({ticker})\n\n"

        result += "#### Valuation Measures\n\n"
        result += "| Metric | Value |\n|--------|-------|\n"
        result += f"| Market Cap | {_fmt(info.get('market_cap'), 'currency')} |\n"
        result += f"| Enterprise Value | {_fmt(info.get('enterprise_value'), 'currency')} |\n"
        result += f"| Trailing P/E | {_fmt(info.get('pe_ratio'), 'ratio')} |\n"
        result += f"| Forward P/E | {_fmt(info.get('forward_pe'), 'ratio')} |\n"
        result += f"| PEG Ratio | {_fmt(info.get('peg_ratio'), 'ratio')} |\n"
        result += f"| Price/Sales | {_fmt(info.get('price_to_sales'), 'ratio')} |\n"
        result += f"| Price/Book | {_fmt(info.get('price_to_book'), 'ratio')} |\n"
        result += "\n"

        result += "#### Financial Highlights\n\n"
        result += "| Metric | Value |\n|--------|-------|\n"
        result += f"| Revenue | {_fmt(info.get('revenue'), 'currency')} |\n"
        result += f"| Gross Profit | {_fmt(info.get('gross_profit'), 'currency')} |\n"
        result += f"| EBITDA | {_fmt(info.get('ebitda'), 'currency')} |\n"
        result += f"| Net Income | {_fmt(info.get('net_income'), 'currency')} |\n"
        result += f"| Diluted EPS | {_fmt(info.get('earnings_per_share'), 'ratio')} |\n"
        result += f"| Profit Margin | {_fmt(info.get('profit_margin'), 'percent')} |\n"
        result += f"| Operating Margin | {_fmt(info.get('operating_margin'), 'percent')} |\n"
        result += f"| ROA | {_fmt(info.get('return_on_assets'), 'percent')} |\n"
        result += f"| ROE | {_fmt(info.get('return_on_equity'), 'percent')} |\n"
        result += "\n"

        result += "#### Trading Information\n\n"
        result += "| Metric | Value |\n|--------|-------|\n"
        result += f"| Current Price | {_fmt(info.get('price'), 'currency')} |\n"
        result += f"| Previous Close | {_fmt(info.get('previous_close'), 'currency')} |\n"
        result += f"| Beta | {_fmt(info.get('beta'), 'ratio')} |\n"
        result += f"| 52-Week High | {_fmt(info.get('fifty_two_week_high'), 'currency')} |\n"
        result += f"| 52-Week Low | {_fmt(info.get('fifty_two_week_low'), 'currency')} |\n"
        result += f"| 50-Day Average | {_fmt(info.get('fifty_day_avg'), 'currency')} |\n"
        result += f"| 200-Day Average | {_fmt(info.get('two_hundred_day_avg'), 'currency')} |\n"
        result += f"| Avg Volume (3mo) | {_fmt(info.get('avg_volume'), 'number')} |\n"
        result += f"| Shares Outstanding | {_fmt(info.get('shares_outstanding'), 'number')} |\n"
        result += f"| Float Shares | {_fmt(info.get('float_shares'), 'number')} |\n"
        result += f"| Short Ratio | {_fmt(info.get('short_ratio'), 'ratio')} |\n"
        result += "\n"

        result += "#### Dividend Info\n\n"
        result += "| Metric | Value |\n|--------|-------|\n"
        result += f"| Dividend Rate | {_fmt(info.get('dividend_rate'), 'currency')} |\n"
        result += f"| Dividend Yield | {_fmt(info.get('dividend_yield'), 'percent')} |\n"
        result += f"| Payout Ratio | {_fmt(info.get('payout_ratio'), 'percent')} |\n"
        result += "\n"

        result += "#### Analyst Targets\n\n"
        result += "| Metric | Value |\n|--------|-------|\n"
        result += f"| Target High | {_fmt(info.get('target_high'), 'currency')} |\n"
        result += f"| Target Low | {_fmt(info.get('target_low'), 'currency')} |\n"
        result += f"| Target Mean | {_fmt(info.get('target_mean'), 'currency')} |\n"
        result += f"| Target Median | {_fmt(info.get('target_median'), 'currency')} |\n"
        result += f"| Recommendation | {info.get('recommendation', 'N/A')} |\n"
        result += f"| Number of Analysts | {info.get('num_analysts', 'N/A')} |\n"
        result += "\n"

        return result
    except Exception as e:
        logger.error(f"Error prefetching stock info for {ticker}: {e}")
        return ""


def prefetch_recommendations(ticker: str) -> str:
    """Prefetch analyst recommendations via yfinance.

    Args:
        ticker: Stock ticker symbol

    Returns:
        Markdown formatted recommendations string, or empty string on error
    """
    try:
        import yfinance as yf
        stock = yf.Ticker(ticker)
        recs = stock.recommendations

        if recs is None or recs.empty:
            logger.warning(f"No recommendations for {ticker}")
            return ""

        return _df_to_markdown(recs, f"Analyst Recommendations: {ticker}")
    except Exception as e:
        logger.error(f"Error prefetching recommendations for {ticker}: {e}")
        return ""


def prefetch_analysis_estimates(ticker: str) -> str:
    """Prefetch earnings/revenue estimates and analyst data via yfinance.

    Replaces firecrawl scrape of Yahoo Finance Analysis page for company_status agent.

    Args:
        ticker: Stock ticker symbol

    Returns:
        Markdown formatted analysis estimates string, or empty string on error
    """
    try:
        import yfinance as yf
        stock = yf.Ticker(ticker)

        result = ""

        # 1. Earnings Estimates
        try:
            earnings_est = stock.earnings_estimate
            if earnings_est is not None and not earnings_est.empty:
                result += _df_to_markdown(earnings_est, f"Earnings Estimates: {ticker}")
                result += "\n"
        except Exception as e:
            logger.debug(f"No earnings estimates for {ticker}: {e}")

        # 2. Revenue Estimates
        try:
            revenue_est = stock.revenue_estimate
            if revenue_est is not None and not revenue_est.empty:
                result += _df_to_markdown(revenue_est, f"Revenue Estimates: {ticker}")
                result += "\n"
        except Exception as e:
            logger.debug(f"No revenue estimates for {ticker}: {e}")

        # 3. EPS Trend
        try:
            eps_trend = stock.eps_trend
            if eps_trend is not None and not eps_trend.empty:
                result += _df_to_markdown(eps_trend, f"EPS Trend: {ticker}")
                result += "\n"
        except Exception as e:
            logger.debug(f"No EPS trend for {ticker}: {e}")

        # 4. EPS Revisions
        try:
            eps_revisions = stock.eps_revisions
            if eps_revisions is not None and not eps_revisions.empty:
                result += _df_to_markdown(eps_revisions, f"EPS Revisions: {ticker}")
                result += "\n"
        except Exception as e:
            logger.debug(f"No EPS revisions for {ticker}: {e}")

        # 5. Growth Estimates
        try:
            growth_est = stock.growth_estimates
            if growth_est is not None and not growth_est.empty:
                result += _df_to_markdown(growth_est, f"Growth Estimates: {ticker}")
                result += "\n"
        except Exception as e:
            logger.debug(f"No growth estimates for {ticker}: {e}")

        # 6. Analyst Price Targets (dict format)
        try:
            targets = stock.analyst_price_targets
            if targets and isinstance(targets, dict):
                result += f"### Analyst Price Targets: {ticker}\n\n"
                result += "| Metric | Value |\n|--------|-------|\n"
                result += f"| Current | ${targets.get('current', 'N/A')} |\n"
                result += f"| High | ${targets.get('high', 'N/A')} |\n"
                result += f"| Low | ${targets.get('low', 'N/A')} |\n"
                result += f"| Mean | ${targets.get('mean', 'N/A')} |\n"
                result += f"| Median | ${targets.get('median', 'N/A')} |\n"
                result += "\n"
        except Exception as e:
            logger.debug(f"No analyst price targets for {ticker}: {e}")

        # 7. Recommendations Summary
        try:
            rec_summary = stock.recommendations_summary
            if rec_summary is not None and not rec_summary.empty:
                result += _df_to_markdown(rec_summary, f"Recommendations Summary: {ticker}")
                result += "\n"
        except Exception as e:
            logger.debug(f"No recommendations summary for {ticker}: {e}")

        if not result:
            logger.warning(f"No analysis estimates data for {ticker}")
            return ""

        return result
    except Exception as e:
        logger.error(f"Error prefetching analysis estimates for {ticker}: {e}")
        return ""


def prefetch_company_profile(ticker: str) -> str:
    """Prefetch company profile data via yfinance.

    Replaces firecrawl profile page scrape for company_overview agent.

    Args:
        ticker: Stock ticker symbol

    Returns:
        Markdown formatted company profile string, or empty string on error
    """
    try:
        import yfinance as yf
        stock = yf.Ticker(ticker)
        info = stock.info

        if not info:
            logger.warning(f"No profile info for {ticker}")
            return ""

        result = f"### Company Profile: {info.get('longName', ticker)}\n\n"

        result += "#### Basic Information\n\n"
        result += "| Field | Value |\n|-------|-------|\n"
        result += f"| Company Name | {info.get('longName', 'N/A')} |\n"
        result += f"| Sector | {info.get('sector', 'N/A')} |\n"
        result += f"| Industry | {info.get('industry', 'N/A')} |\n"
        result += f"| Website | {info.get('website', 'N/A')} |\n"
        employees = info.get('fullTimeEmployees')
        result += f"| Full-Time Employees | {employees:,} |\n" if employees else "| Full-Time Employees | N/A |\n"
        city = info.get('city', '')
        state = info.get('state', '')
        country = info.get('country', '')
        address = ", ".join(filter(None, [city, state, country]))
        result += f"| Headquarters | {address or 'N/A'} |\n"
        result += "\n"

        description = info.get('longBusinessSummary', '')
        if description:
            result += "#### Business Description\n\n"
            result += f"{description}\n\n"

        officers = info.get('companyOfficers', [])
        if officers:
            result += "#### Key Executives\n\n"
            result += "| Name | Title | Total Pay |\n|------|-------|-----------|\n"
            for officer in officers[:10]:
                name = officer.get('name', 'N/A')
                title = officer.get('title', 'N/A')
                pay = officer.get('totalPay', 0)
                pay_str = f"${pay:,.0f}" if pay else "N/A"
                result += f"| {name} | {title} | {pay_str} |\n"
            result += "\n"

        return result
    except Exception as e:
        logger.error(f"Error prefetching company profile for {ticker}: {e}")
        return ""


def prefetch_financial_statements(ticker: str) -> str:
    """Prefetch financial statements (income statement, balance sheet, cash flow) via yfinance.

    Replaces SEC EDGAR get_financials/get_key_metrics calls.

    Args:
        ticker: Stock ticker symbol

    Returns:
        Markdown formatted financial statements string, or empty string on error
    """
    try:
        import yfinance as yf
        stock = yf.Ticker(ticker)

        result = ""

        # Annual income statement
        try:
            income = stock.income_stmt
            if income is not None and not income.empty:
                result += _df_to_markdown(income, f"Annual Income Statement: {ticker}")
                result += "\n"
        except Exception as e:
            logger.debug(f"No annual income statement for {ticker}: {e}")

        # Annual balance sheet
        try:
            balance = stock.balance_sheet
            if balance is not None and not balance.empty:
                result += _df_to_markdown(balance, f"Annual Balance Sheet: {ticker}")
                result += "\n"
        except Exception as e:
            logger.debug(f"No annual balance sheet for {ticker}: {e}")

        # Annual cash flow
        try:
            cashflow = stock.cashflow
            if cashflow is not None and not cashflow.empty:
                result += _df_to_markdown(cashflow, f"Annual Cash Flow: {ticker}")
                result += "\n"
        except Exception as e:
            logger.debug(f"No annual cash flow for {ticker}: {e}")

        # Quarterly income statement (latest 4 quarters)
        try:
            q_income = stock.quarterly_income_stmt
            if q_income is not None and not q_income.empty:
                result += _df_to_markdown(q_income, f"Quarterly Income Statement: {ticker}")
                result += "\n"
        except Exception as e:
            logger.debug(f"No quarterly income statement for {ticker}: {e}")

        # Quarterly balance sheet
        try:
            q_balance = stock.quarterly_balance_sheet
            if q_balance is not None and not q_balance.empty:
                result += _df_to_markdown(q_balance, f"Quarterly Balance Sheet: {ticker}")
                result += "\n"
        except Exception as e:
            logger.debug(f"No quarterly balance sheet for {ticker}: {e}")

        # Quarterly cash flow
        try:
            q_cashflow = stock.quarterly_cashflow
            if q_cashflow is not None and not q_cashflow.empty:
                result += _df_to_markdown(q_cashflow, f"Quarterly Cash Flow: {ticker}")
                result += "\n"
        except Exception as e:
            logger.debug(f"No quarterly cash flow for {ticker}: {e}")

        if not result:
            logger.warning(f"No financial statements for {ticker}")
            return ""

        return result
    except Exception as e:
        logger.error(f"Error prefetching financial statements for {ticker}: {e}")
        return ""


def _clean_xbrl_label(raw: str) -> str:
    """Clean XBRL member names into readable labels.

    Args:
        raw: Raw XBRL member name (e.g., "IPhoneMember", "GreaterChinaSegmentMember")

    Returns:
        Human-readable label (e.g., "iPhone", "Greater China")
    """
    import re
    label = raw.replace("Member", "").replace("Segment", "")

    known = {
        "IPhone": "iPhone", "IPad": "iPad", "IMac": "iMac",
        "WearablesHomeandAccessories": "Wearables, Home & Accessories",
        "WearablesHomeAndAccessories": "Wearables, Home & Accessories",
        "ServiceOther": "Services & Other",
        "OEMAndOther": "OEM & Other",
        "LinkedInCorporation": "LinkedIn",
        "MicrosoftThreeSixFiveCommercialProductsAndCloudServices": "Microsoft 365 Commercial",
        "MicrosoftThreeSixFiveConsumerProductsAndCloudServices": "Microsoft 365 Consumer",
        "ServerProductsAndCloudServices": "Server Products & Cloud Services",
        "SearchAndNewsAdvertising": "Search & News Advertising",
        "DynamicsProductsAndCloudServices": "Dynamics Products & Cloud Services",
        "EnterpriseAndPartnerServices": "Enterprise & Partner Services",
        "WindowsAndDevices": "Windows & Devices",
        "OtherProductsAndServices": "Other Products & Services",
        "ComputeAndNetworking": "Compute & Networking",
        "ProfessionalVisualization": "Professional Visualization",
        "RestOfAsiaPacific": "Rest of Asia Pacific",
        "GreaterChina": "Greater China",
        "OtherCountries": "Other Countries",
        "NonUs": "International",
    }
    if label in known:
        return known[label]

    # CamelCase to spaces
    label = re.sub(r'([a-z])([A-Z])', r'\1 \2', label)
    label = re.sub(r'([A-Z]+)([A-Z][a-z])', r'\1 \2', label)
    return label.strip()


def _parse_10k_segment_revenue(html_content: str) -> str:
    """Parse segment revenue data from 10-K XBRL inline HTML.

    Extracts Products/Services split, product line breakdown,
    geographic segment revenue, and country-level revenue from
    XBRL inline tags embedded in SEC 10-K filings.

    Args:
        html_content: Raw HTML content of 10-K filing

    Returns:
        Markdown formatted segment revenue data, or empty string
    """
    import re

    # 1. Parse all XBRL contexts (dimension definitions)
    context_pattern = r'<xbrli:context[^>]*id="([^"]+)"[^>]*>(.*?)</xbrli:context>'
    contexts = {}
    for m in re.finditer(context_pattern, html_content, re.DOTALL):
        ctx_id = m.group(1)
        ctx_body = m.group(2)
        dims = re.findall(
            r'<xbrldi:explicitMember[^>]*dimension="([^"]+)"[^>]*>([^<]+)', ctx_body
        )
        periods = re.findall(r'<xbrli:(?:startDate|endDate)>([^<]+)', ctx_body)
        period = f"{periods[0]}~{periods[1]}" if len(periods) >= 2 else ""
        contexts[ctx_id] = {"dims": dict(dims), "period": period}

    # 2. Parse all revenue XBRL tags
    rev_pattern = (
        r'<ix:nonFraction[^>]*contextRef="([^"]+)"[^>]*'
        r'name="[^"]*Revenue[^"]*"[^>]*>([^<]+)</ix:nonFraction>'
    )
    revenues = []
    for m in re.finditer(rev_pattern, html_content):
        ctx_id = m.group(1)
        value_str = m.group(2).strip().replace(',', '')
        try:
            value = int(value_str)
        except ValueError:
            continue
        ctx = contexts.get(ctx_id, {"dims": {}, "period": ""})
        revenues.append({"value": value, "dims": ctx["dims"], "period": ctx["period"]})

    if not revenues:
        return ""

    # 3. Categorize revenues by dimension type
    product_service = {}  # {(category, period): value}
    product_lines = {}    # {(product, period): value}
    geographic = {}       # {(region, period): value}
    geo_country = {}      # {(country, period): value}
    totals = {}           # {period: value}

    for r in revenues:
        dims = r["dims"]
        period = r["period"]
        value = r["value"]

        if not period or value < 5:
            continue

        seg_axis = dims.get("us-gaap:StatementBusinessSegmentsAxis", "")
        product_axis = dims.get("srt:ProductOrServiceAxis", "")
        consol_axis = dims.get("srt:ConsolidationItemsAxis", "")
        geo_axis = dims.get("srt:StatementGeographicalAxis", "")

        if seg_axis and consol_axis:
            region = seg_axis.split(":")[-1].replace("SegmentMember", "").replace("Member", "")
            geographic[(region, period)] = value
        elif product_axis:
            product = product_axis.split(":")[-1].replace("Member", "")
            if product in ("Product", "Service"):
                product_service[(product, period)] = value
            else:
                product_lines[(product, period)] = value
        elif geo_axis:
            country = geo_axis.split(":")[-1].replace("Member", "")
            geo_country[(country, period)] = value
        elif not dims:
            if period not in totals or value > totals.get(period, 0):
                totals[period] = value

    # 4. Format as markdown
    result = "### Segment Revenue Data (from 10-K filing, in millions USD)\n\n"

    all_periods = sorted(set(
        p for _, p in list(product_service.keys()) + list(product_lines.keys()) +
        list(geographic.keys()) + list(geo_country.keys())
    ), reverse=True)

    if not all_periods:
        return ""

    def _period_label(p):
        parts = p.split("~")
        return f"FY{parts[1][:4]}" if len(parts) == 2 else p

    def _fmt_value(v):
        if v >= 1000:
            return f"${v / 1000:.1f}B"
        return f"${v:,}M"

    # Products vs Services (skip if Services is zero)
    has_services = any(v > 0 for (k, _), v in product_service.items() if k == "Service")
    if product_service and has_services:
        periods = sorted(set(p for _, p in product_service.keys()), reverse=True)
        cols = [_period_label(p) for p in periods]
        result += "#### Revenue by Category\n\n"
        result += "| Category | " + " | ".join(cols) + " |\n"
        result += "|----------|" + "|".join(["--------"] * len(cols)) + "|\n"
        for seg_type in ["Product", "Service"]:
            vals = [_fmt_value(product_service.get((seg_type, p), 0)) for p in periods]
            result += f"| {seg_type}s | " + " | ".join(vals) + " |\n"
        total_vals = [_fmt_value(totals.get(p, 0)) for p in periods]
        result += "| **Total** | " + " | ".join(total_vals) + " |\n\n"

    # Product lines
    if product_lines:
        periods = sorted(set(p for _, p in product_lines.keys()), reverse=True)
        cols = [_period_label(p) for p in periods]
        result += "#### Revenue by Product Line\n\n"
        result += "| Product | " + " | ".join(cols) + " |\n"
        result += "|---------|" + "|".join(["--------"] * len(cols)) + "|\n"
        products = list(set(prod for prod, _ in product_lines.keys()))
        latest = periods[0] if periods else ""
        products.sort(key=lambda x: product_lines.get((x, latest), 0), reverse=True)
        products = products[:8]  # Limit to top 8 to control token usage
        for prod in products:
            label = _clean_xbrl_label(prod)
            vals = [_fmt_value(product_lines.get((prod, p), 0)) for p in periods]
            result += f"| {label} | " + " | ".join(vals) + " |\n"
        result += "\n"

    # Geographic segments
    if geographic:
        periods = sorted(set(p for _, p in geographic.keys()), reverse=True)
        cols = [_period_label(p) for p in periods]
        result += "#### Revenue by Geographic Segment\n\n"
        result += "| Region | " + " | ".join(cols) + " |\n"
        result += "|--------|" + "|".join(["--------"] * len(cols)) + "|\n"
        regions = list(set(r for r, _ in geographic.keys()))
        latest = periods[0] if periods else ""
        regions.sort(key=lambda x: geographic.get((x, latest), 0), reverse=True)
        for region in regions:
            label = _clean_xbrl_label(region)
            vals = [_fmt_value(geographic.get((region, p), 0)) for p in periods]
            result += f"| {label} | " + " | ".join(vals) + " |\n"
        total_vals = [_fmt_value(totals.get(p, 0)) for p in periods]
        result += "| **Total** | " + " | ".join(total_vals) + " |\n\n"

    # Country-level (skip if geographic segments already provide regional breakdown)
    if geo_country and not geographic:
        periods = sorted(set(p for _, p in geo_country.keys()), reverse=True)
        cols = [_period_label(p) for p in periods]
        result += "#### Revenue by Country\n\n"
        result += "| Country | " + " | ".join(cols) + " |\n"
        result += "|---------|" + "|".join(["--------"] * len(cols)) + "|\n"
        countries = list(set(c for c, _ in geo_country.keys()))
        latest = periods[0] if periods else ""
        countries.sort(key=lambda x: geo_country.get((x, latest), 0), reverse=True)
        for country in countries:
            label = _clean_xbrl_label(country)
            vals = [_fmt_value(geo_country.get((country, p), 0)) for p in periods]
            result += f"| {label} | " + " | ".join(vals) + " |\n"
        result += "\n"

    return result


def prefetch_segment_revenue(ticker: str) -> str:
    """Prefetch segment revenue data from latest 10-K filing via Yahoo Finance CDN.

    Uses yfinance sec_filings to find 10-K URL, downloads XBRL inline HTML from
    cdn.yahoofinance.com, and parses segment revenue breakdowns.

    Args:
        ticker: Stock ticker symbol

    Returns:
        Markdown formatted segment revenue data, or empty string on error
    """
    try:
        import yfinance as yf
        import urllib.request

        stock = yf.Ticker(ticker)
        filings = stock.sec_filings

        if not filings:
            logger.warning(f"No SEC filings for {ticker}")
            return ""

        # Find latest 10-K or 10-Q (whichever is most recent)
        # sec_filings are returned in date-descending order
        filing = None
        for f in filings:
            ftype = f.get('type', '')
            if ftype in ('10-K', '10-Q'):
                filing = f
                break

        if not filing:
            logger.warning(f"No 10-K/10-Q filing found for {ticker}")
            return ""

        filing_type = filing.get('type', '10-K')
        url = filing.get('exhibits', {}).get(filing_type, '')
        if not url:
            logger.warning(f"No {filing_type} exhibit URL for {ticker}")
            return ""

        # Download filing HTML from Yahoo Finance CDN
        req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
        resp = urllib.request.urlopen(req, timeout=30)
        html_content = resp.read().decode('utf-8', errors='replace')

        if not html_content:
            logger.warning(f"Empty {filing_type} HTML for {ticker}")
            return ""

        result = _parse_10k_segment_revenue(html_content)
        if result:
            # Update title to reflect actual filing type
            result = result.replace("from 10-K filing", f"from {filing_type} filing")
            logger.info(f"Parsed segment revenue for {ticker} from {filing_type} ({len(html_content):,} chars HTML)")
        else:
            logger.warning(f"No segment revenue data found in {filing_type} for {ticker}")

        return result
    except Exception as e:
        logger.warning(f"Segment revenue prefetch failed for {ticker}: {e}")
        return ""


def prefetch_us_analysis_data(ticker: str) -> dict:
    """Prefetch all data needed for US stock analysis agents.

    Args:
        ticker: Stock ticker symbol (e.g., "AAPL")

    Returns:
        Dictionary with prefetched data:
        - "stock_ohlcv": OHLCV data as markdown
        - "holder_info": Institutional holder data as markdown
        - "market_indices": Dict of index data
    """
    result = {}

    # 1. Stock OHLCV
    stock_ohlcv = prefetch_us_stock_ohlcv(ticker, period="1y")
    if stock_ohlcv:
        result["stock_ohlcv"] = stock_ohlcv

    # 2. Holder info
    holder_info = prefetch_us_holder_info(ticker)
    if holder_info:
        result["holder_info"] = holder_info

    # 3. Market indices
    market_indices = prefetch_us_market_indices()
    if market_indices:
        result["market_indices"] = market_indices

    # 4. Stock info (for company_status - replaces key-statistics/financials firecrawl + yahoo_finance MCP)
    stock_info = prefetch_stock_info(ticker)
    if stock_info:
        result["stock_info"] = stock_info

    # 5. Recommendations (for company_status - replaces yahoo_finance MCP get_recommendations)
    recommendations = prefetch_recommendations(ticker)
    if recommendations:
        result["recommendations"] = recommendations

    # 6. Company profile (for company_overview - replaces firecrawl profile page)
    company_profile = prefetch_company_profile(ticker)
    if company_profile:
        result["company_profile"] = company_profile

    # 7. Analysis estimates (for company_status - replaces firecrawl Analysis page)
    analysis_estimates = prefetch_analysis_estimates(ticker)
    if analysis_estimates:
        result["analysis_estimates"] = analysis_estimates

    # 8. Financial statements (for company_status - replaces SEC EDGAR financials)
    financial_statements = prefetch_financial_statements(ticker)
    if financial_statements:
        result["financial_statements"] = financial_statements

    # 9. Segment revenue (for company_overview - parsed from 10-K XBRL via Yahoo Finance CDN)
    segment_revenue = prefetch_segment_revenue(ticker)
    if segment_revenue:
        result["segment_revenue"] = segment_revenue

    if result:
        logger.info(f"Prefetched US data for {ticker}: {list(result.keys())}")
    else:
        logger.warning(f"Failed to prefetch any US data for {ticker}")

    return result


def _log_regime_snapshot(market: str, computed: dict) -> None:
    """Append a regime snapshot to logs/regime_history.jsonl for distribution analysis.

    사이클당 1회 기록 → 운영에서 regime 분포/휩쏘 관측용. 실패해도 무해(파이프라인 영향 0).
    """
    try:
        if not computed:
            return
        import json as _json
        import os as _os
        from datetime import datetime as _dt
        rec = {
            "ts": _dt.now().strftime("%Y-%m-%d %H:%M:%S"),
            "market": market,
            "regime": computed.get("market_regime"),
            "confidence": computed.get("regime_confidence"),
        }
        s = computed.get("index_summary") or {}
        for k in ("sp500_vs_50d_ma", "sp500_vs_200d_ma", "sp500_ma_50_200_cross",
                  "sp500_4w_change_pct", "vix_level",
                  "kospi_vs_60d_ma", "kospi_vs_120d_ma", "kospi_ma_60_120_cross",
                  "kospi_2w_change_pct",
                  "highvol_override_mode", "highvol_drawdown_override",
                  "distribution_days"):
            if k in s:
                rec[k] = s[k]
        log_dir = _os.path.join(_os.getcwd(), "logs")
        _os.makedirs(log_dir, exist_ok=True)
        with open(_os.path.join(log_dir, "regime_history.jsonl"), "a") as f:
            f.write(_json.dumps(rec, ensure_ascii=False) + "\n")
        logger.info(f"[regime] {market}: {rec['regime']} (conf {rec['confidence']})")
    except Exception as e:
        logger.warning(f"[regime] snapshot log failed: {e}")


def prefetch_us_macro_intelligence_data(reference_date: str = None) -> dict:
    """Prefetch data for US macro intelligence analysis.

    Fetches S&P 500, NASDAQ, VIX index data, then computes market regime
    programmatically from price data (not LLM-based).

    Args:
        reference_date: Analysis date (YYYYMMDD) - used for logging

    Returns:
        Dictionary with:
        - "sp500_ohlcv_md": S&P 500 20-day OHLCV as markdown
        - "nasdaq_ohlcv_md": NASDAQ 20-day OHLCV as markdown
        - "vix_ohlcv_md": VIX 20-day OHLCV as markdown
        - "computed_regime": programmatically computed regime info dict
    """
    result = {}

    try:
        client = _get_us_data_client()
    except Exception as e:
        logger.error(f"Failed to get US data client: {e}")
        return result

    # 1. S&P 500
    sp500_df = None
    try:
        sp500_df = client.get_ohlcv("^GSPC", period="1y", interval="1d")
        if sp500_df is not None and not sp500_df.empty:
            sp500_20d = sp500_df.tail(20)
            sp500_20d.columns = [col.title().replace("_", " ") for col in sp500_20d.columns]
            sp500_20d.index.name = "Date"
            result["sp500_ohlcv_md"] = _df_to_markdown(sp500_20d, "S&P 500 (20 trading days)")
    except Exception as e:
        logger.error(f"Error fetching S&P 500: {e}")

    # 2. NASDAQ
    nasdaq_df = None
    try:
        nasdaq_df = client.get_ohlcv("^IXIC", period="1y", interval="1d")
        if nasdaq_df is not None and not nasdaq_df.empty:
            nasdaq_20d = nasdaq_df.tail(20)
            nasdaq_20d.columns = [col.title().replace("_", " ") for col in nasdaq_20d.columns]
            nasdaq_20d.index.name = "Date"
            result["nasdaq_ohlcv_md"] = _df_to_markdown(nasdaq_20d, "NASDAQ Composite (20 trading days)")
    except Exception as e:
        logger.error(f"Error fetching NASDAQ: {e}")

    # 3. VIX
    vix_df = None
    try:
        vix_df = client.get_ohlcv("^VIX", period="1y", interval="1d")
        if vix_df is not None and not vix_df.empty:
            vix_20d = vix_df.tail(20)
            vix_20d.columns = [col.title().replace("_", " ") for col in vix_20d.columns]
            vix_20d.index.name = "Date"
            result["vix_ohlcv_md"] = _df_to_markdown(vix_20d, "VIX (20 trading days)")
    except Exception as e:
        logger.error(f"Error fetching VIX: {e}")

    # 4. Compute regime programmatically
    if sp500_df is not None and not sp500_df.empty:
        result["computed_regime"] = _compute_us_regime(sp500_df, nasdaq_df, vix_df)
        _log_regime_snapshot("US", result["computed_regime"])

    if result:
        logger.info(f"Prefetched US macro intelligence data: {list(result.keys())}")

    return result


# --- O'Neil Distribution Day (deterministic, 정보 주입 전용) ----------------
# 설계 결정(tasks/distribution_day_design.md): 분산일은 결정론적으로 '계산'해 index_summary에
# 정보로만 주입하고, regime의 기계적 강등은 하지 않는다. (강등은 매수+매도 양쪽을 뒤집어
# US melt-up에서 조기청산 손실을 유발했고, 시장별 임계는 과최적화 위험이 컸다.) 분산일을
# 어떻게 가중할지(신규매수 보수화 등)는 프롬프트에서 LLM이 판단한다 — O'Neil 본래의 재량적 용법.
# 분산일 파라미터 (O'Neil/IBD). drop=-0.2% 종가, 거래량 전일 초과, 25거래일 윈도우, +5% 회복 만료.
DISTRIBUTION_WINDOW = 25
DISTRIBUTION_DROP_PCT = 0.2
DISTRIBUTION_RECOVERY_PCT = 5.0

# 고변동·낙폭 override 파라미터 (TUNABLE — 배포 전 tools/regime_backtest.py 로 검증/튜닝 권장).
# KR cores/data_prefetch.py 와 동일 구현(포크 병렬). 장기이평(50/200일선) 위에 '가격 레벨이
# 지연되어' 떠 있으나 실제로는 급락형 고변동(whipsaw) 국면이 strong/moderate_bull 로 관대하게
# 분류돼 고점매수→손절이 반복되는 것을 방지. 낙폭 조건으로 melt-up(신고가 부근 고변동)은 배제
# → 989줄에서 우려한 'melt-up 조기청산' 문제를 구조적으로 회피(과최적화 경계 존중).
# 참고: US 는 VIX 를 이미 strong_bull/strong_bear 조건에 쓰지만 moderate_bull 분기엔 없어
# 동일 사각지대가 존재 → 여기서 실현변동성+낙폭으로 보강(시장 무관 대칭 구현).
HIVOL_DD_VOL_PCT = 2.5       # 최근 10일 일간수익률 표준편차 임계(%). S&P500 평시 ~1%.
HIVOL_DD_DRAWDOWN_PCT = 8.0  # 최근 20일 고점 대비 낙폭 임계(%).
HIVOL_DD_NET_DECLINE_PCT = -3.0  # 최근 10일 '순변화' 임계(%). 급등형(net +) 배제 핵심조건.
HIVOL_DD_CONFIDENCE = 0.55   # 강등 시 신뢰도(낮춤).
_BULL_REGIMES = ("strong_bull", "moderate_bull")


def _high_vol_drawdown_override(closes, regime: str, confidence: float):
    """급락형 고변동 국면이면 bull 라벨을 sideways 로 강등(순수 함수, 부작용 없음).

    발동 조건(모두 충족): regime 이 bull 계열 AND 최근 10일 실현변동성 >= HIVOL_DD_VOL_PCT
    AND 최근 20일 고점 대비 낙폭 >= HIVOL_DD_DRAWDOWN_PCT (= melt-up 이 아님).
    강등은 sideways 까지만 — 기존 REGIME 매수 문턱/슬롯/가중치가 자동 보수화된다.

    Returns: (regime, confidence, reason) — 미발동 시 reason 은 None.
    """
    if regime not in _BULL_REGIMES:
        return regime, confidence, None
    if closes is None or len(closes) < 11:
        return regime, confidence, None
    arr = np.asarray(closes, dtype=float)
    arr = arr[~np.isnan(arr)]
    if len(arr) < 11:
        return regime, confidence, None
    recent = arr[-11:]
    rets = np.diff(recent) / recent[:-1]
    vol_pct = float(np.std(rets) * 100)
    net_change_pct = (float(arr[-1]) - float(recent[0])) / float(recent[0]) * 100  # 최근 순변화
    window = arr[-20:] if len(arr) >= 20 else arr
    peak = float(np.max(window))
    drawdown_pct = (peak - float(arr[-1])) / peak * 100 if peak > 0 else 0.0
    # 3중 조건: 고변동 AND 고점대비 낙폭 AND '실제로 하락 중'(순변화 마이너스).
    # 순변화 조건이 급등형(net +)/횡보 고변동을 배제 → melt-up 오강등 방지 핵심.
    if (vol_pct >= HIVOL_DD_VOL_PCT
            and drawdown_pct >= HIVOL_DD_DRAWDOWN_PCT
            and net_change_pct <= HIVOL_DD_NET_DECLINE_PCT):
        reason = (
            f"{regime}->sideways (vol10d={vol_pct:.1f}%>={HIVOL_DD_VOL_PCT}, "
            f"dd20d={drawdown_pct:.1f}%>={HIVOL_DD_DRAWDOWN_PCT}, "
            f"net10d={net_change_pct:.1f}%<={HIVOL_DD_NET_DECLINE_PCT})"
        )
        return "sideways", min(confidence, HIVOL_DD_CONFIDENCE), reason
    return regime, confidence, None


def _count_distribution_days(df, close_col, volume_col=None,
                             window: int = DISTRIBUTION_WINDOW,
                             drop_threshold_pct: float = DISTRIBUTION_DROP_PCT,
                             recovery_pct: float = DISTRIBUTION_RECOVERY_PCT):
    """O'Neil 분산일 카운트 (결정론적).

    분산일 = 지수가 전일 종가 대비 >= drop_threshold_pct% 하락 마감 AND 거래량이 전일 초과.
    만료: (1) window 거래일 경과 시 윈도우 밖으로 자동 제외, (2) 분산일 이후 어떤 종가가
    그 분산일 종가 대비 +recovery_pct% 이상 상승하면 카운트에서 제거.

    Returns:
        {"count": int, "window": int, "raw_count": int} 또는 거래량 불가 시 None.
    """
    try:
        d = df.sort_index()
        if volume_col is None:
            for c in ["Volume", "거래량", "volume"]:
                if c in d.columns:
                    volume_col = c
                    break
        if volume_col is None or close_col not in d.columns:
            return None
        closes = d[close_col].astype(float).values
        vols = d[volume_col].astype(float).values
        n = len(closes)
        if n < 2:
            return None
        import math as _math
        valid_vol = [v for v in vols if not _math.isnan(v) and v > 0]
        if not valid_vol:
            return None
        start = max(1, n - window)
        raw = 0
        kept = 0
        running_max_after = -1.0
        flags = []
        for i in range(n - 1, start - 1, -1):
            prev_c = closes[i - 1]
            cur_c = closes[i]
            if prev_c <= 0:
                flags.append((i, False, running_max_after))
                running_max_after = max(running_max_after, cur_c)
                continue
            pct = (cur_c - prev_c) / prev_c * 100.0
            vol_up = vols[i] > vols[i - 1]
            is_dist = (pct <= -drop_threshold_pct) and vol_up
            flags.append((i, is_dist, running_max_after))
            running_max_after = max(running_max_after, cur_c)
        for (i, is_dist, max_after) in flags:
            if not is_dist:
                continue
            raw += 1
            if max_after >= closes[i] * (1 + recovery_pct / 100.0):
                continue
            kept += 1
        return {"count": kept, "window": window, "raw_count": raw}
    except Exception:
        return None


def _inject_distribution_days(index_summary, df, close_col) -> None:
    """분산일 카운트를 결정론적으로 계산해 index_summary에 정보로 주입(강등 없음).

    거래량 결측/불가 시 distribution_days=None. regime/confidence는 변경하지 않는다.
    """
    dist = _count_distribution_days(df, close_col)
    index_summary["distribution_window"] = DISTRIBUTION_WINDOW
    index_summary["distribution_days"] = None if dist is None else dist["count"]


def _compute_us_regime(sp500_df: pd.DataFrame, nasdaq_df: pd.DataFrame = None, vix_df: pd.DataFrame = None) -> dict:
    """Compute US market regime programmatically from index data.

    Uses S&P 500 20-day MA, 4-week change, and VIX level for classification.

    Returns:
        Dict with regime classification, index summary, and confidence.
    """
    if sp500_df is None or sp500_df.empty or len(sp500_df) < 10:
        return {"market_regime": "sideways", "regime_confidence": 0.3, "simple_ma_regime": "sideways"}

    sp500_df = sp500_df.sort_index()
    df_20d = sp500_df.tail(20)

    # Find close column
    close_col = None
    for col_name in ["Close", "close", "Adj Close"]:
        if col_name in sp500_df.columns:
            close_col = col_name
            break
    if not close_col:
        close_col = sp500_df.columns[3]

    current_price = float(df_20d[close_col].iloc[-1])
    ma_20d = float(df_20d[close_col].mean())

    # 4-week change (20 trading days)
    if len(df_20d) >= 20:
        price_4w_ago = float(df_20d[close_col].iloc[0])
    elif len(df_20d) >= 10:
        price_4w_ago = float(df_20d[close_col].iloc[0])
    else:
        price_4w_ago = float(df_20d[close_col].iloc[0])
    change_4w_pct = ((current_price - price_4w_ago) / price_4w_ago) * 100

    # MA position (short-term, 20-day)
    ma_diff_pct = ((current_price - ma_20d) / ma_20d) * 100
    above_ma = current_price > ma_20d

    # Trend MAs (50/200) from FULL history — needs ~1y fetch.
    # O'Neil/Minervini trend template: 200MA = primary bull/bear divider, 50MA = intermediate.
    closes_full = sp500_df[close_col].dropna()
    ma_50 = float(closes_full.tail(50).mean()) if len(closes_full) >= 50 else None
    ma_200 = float(closes_full.tail(200).mean()) if len(closes_full) >= 200 else None
    above_50 = (current_price > ma_50) if ma_50 is not None else None
    above_200 = (current_price > ma_200) if ma_200 is not None else None
    # golden = 50MA above 200MA (bullish alignment); False = dead cross
    golden = (ma_50 > ma_200) if (ma_50 is not None and ma_200 is not None) else None

    # VIX level
    vix_current = None
    vix_level = "moderate"
    if vix_df is not None and not vix_df.empty:
        vix_close = None
        for col_name in ["Close", "close", "Adj Close"]:
            if col_name in vix_df.columns:
                vix_close = col_name
                break
        if vix_close:
            vix_current = float(vix_df[vix_close].iloc[-1])
            if vix_current < 15:
                vix_level = "low"
            elif vix_current < 20:
                vix_level = "moderate"
            elif vix_current < 25:
                vix_level = "elevated"
            else:
                vix_level = "high"

    # simple_ma_regime
    if abs(ma_diff_pct) <= 0.5:
        simple_ma_regime = "sideways"
    elif above_ma:
        simple_ma_regime = "bull"
    else:
        simple_ma_regime = "bear"

    # S&P 500 20d trend
    if change_4w_pct > 2:
        sp500_trend = "up"
    elif change_4w_pct < -2:
        sp500_trend = "down"
    else:
        sp500_trend = "sideways"

    # NASDAQ trend
    nasdaq_trend = "sideways"
    if nasdaq_df is not None and not nasdaq_df.empty:
        try:
            nasdaq_df = nasdaq_df.sort_index()
            nd_20d = nasdaq_df.tail(20)
            nd_close = None
            for col_name in ["Close", "close", "Adj Close"]:
                if col_name in nasdaq_df.columns:
                    nd_close = col_name
                    break
            if nd_close and len(nd_20d) >= 10:
                nd_current = float(nd_20d[nd_close].iloc[-1])
                nd_prev = float(nd_20d[nd_close].iloc[0])
                nd_change = ((nd_current - nd_prev) / nd_prev) * 100
                if nd_change > 2:
                    nasdaq_trend = "up"
                elif nd_change < -2:
                    nasdaq_trend = "down"
        except Exception:
            pass

    # Market regime classification.
    # Trend-template (200MA primary divider) when 200MA available; else legacy 20MA logic.
    # Output strings unchanged (6 regimes) so downstream buy-matrix/prompts stay compatible.
    if ma_200 is not None:
        if above_200:
            # Primary uptrend (price above 200MA)
            if above_50 and golden and change_4w_pct > 3 and vix_level in ("low", "moderate"):
                regime = "strong_bull"
                confidence = 0.90
            elif above_50 or change_4w_pct >= 0:
                regime = "moderate_bull"
                confidence = 0.78
            else:
                # above primary trend but short-term pullback below 50MA
                regime = "sideways"
                confidence = 0.62
        else:
            # Primary downtrend (price below 200MA) → never 'bull'. Bear-rally protection:
            # a sharp 4w bounce here is classified cautious (sideways), not strong_bull.
            if golden is False and change_4w_pct < -5 and vix_level in ("elevated", "high"):
                regime = "strong_bear"
                confidence = 0.90
            elif change_4w_pct < 0:
                regime = "moderate_bear"
                confidence = 0.78
            else:
                regime = "sideways"
                confidence = 0.55
    else:
        # Legacy 20MA logic (insufficient history for 200MA) — backward compatible
        if above_ma and change_4w_pct > 3 and vix_level in ("low", "moderate"):
            regime = "strong_bull"
            confidence = 0.85
        elif above_ma and change_4w_pct >= 0:
            regime = "moderate_bull"
            confidence = 0.75
        elif abs(ma_diff_pct) <= 1 and abs(change_4w_pct) < 2:
            regime = "sideways"
            confidence = 0.65
        elif not above_ma and change_4w_pct < -5 and vix_level in ("elevated", "high"):
            regime = "strong_bear"
            confidence = 0.85
        else:
            regime = "moderate_bear"
            confidence = 0.75

    index_summary = {
        "sp500_20d_trend": sp500_trend,
        "sp500_vs_20d_ma": "above" if above_ma else "below",
        "sp500_4w_change_pct": round(change_4w_pct, 2),
        "sp500_current": round(current_price, 2),
        "sp500_20d_ma": round(ma_20d, 2),
        "nasdaq_20d_trend": nasdaq_trend,
    }
    # Trend MA fields (additive — present only when enough history)
    if ma_50 is not None:
        index_summary["sp500_50d_ma"] = round(ma_50, 2)
        index_summary["sp500_vs_50d_ma"] = "above" if above_50 else "below"
    if ma_200 is not None:
        index_summary["sp500_200d_ma"] = round(ma_200, 2)
        index_summary["sp500_vs_200d_ma"] = "above" if above_200 else "below"
    if golden is not None:
        index_summary["sp500_ma_50_200_cross"] = "golden" if golden else "dead"
    if vix_current is not None:
        index_summary["vix_current"] = round(vix_current, 2)
        index_summary["vix_level"] = vix_level

    # O'Neil 분산일 결정론 카운트를 index_summary에 정보로 주입(강등 없음 — LLM이 프롬프트에서 판단)
    _inject_distribution_days(index_summary, sp500_df, close_col)

    # 고변동·낙폭 override (모드: shadow[기본]/active/off). KR 과 동일 시맨틱.
    #   shadow = 강등 '판단'만 계산·로깅, regime 은 그대로(매매 무영향, 관찰용).
    #   active = 실제 강등 적용.  off = 완전 비활성.
    import os as _os
    _mode = _os.environ.get("REGIME_HIVOL_OVERRIDE", "shadow").strip().lower()
    index_summary["highvol_override_mode"] = _mode
    index_summary["highvol_drawdown_override"] = None
    if _mode != "off":
        _new_regime, _new_conf, _reason = _high_vol_drawdown_override(
            closes_full.to_numpy(), regime, confidence
        )
        index_summary["highvol_drawdown_override"] = _reason
        if _reason:
            if _mode == "active":
                logger.info(f"[regime] US HIVOL override ACTIVE: {_reason}")
                regime, confidence = _new_regime, _new_conf
            else:
                logger.info(f"[regime] US HIVOL override SHADOW(관찰,미적용): would {_reason}")

    return {
        "market_regime": regime,
        "regime_confidence": confidence,
        "simple_ma_regime": simple_ma_regime,
        "index_summary": index_summary,
    }
