"""
Market filters on top of FinLab's verified stock-selection strategies (FinLab research, 2026-11).

Each base strategy is the strategy page's own downloadable script, run unchanged. We record the
positions and settings it sends to finlab's sim(), then put one of two filters on top:

  trend      Filter A, market moving average (Faber 2007). When the market closes below its
             N-day average times (1 - buffer), hold nothing; hold the base strategy's current
             stocks again once the market closes back at or above the average. Taiwan uses the
             TAIEX total return index, the US uses QQQ with dividends.
             Taiwan N 60/120/240, US N 100/150/200, buffer 0/2%/4%.
  vol        Filter B, volatility management (Moreira & Muir 2017, Barroso & Santa-Clara 2015).
             At each month's last close, exposure = min(1, target / sigma), where sigma is the
             base strategy's realised volatility over the last L days and target is the median
             of sigma up to that day times m. The rest is cash until the next month.
             L 21/63/126 days x m 0.8/1.0/1.2.

The original setting is the middle of each grid. Signals use a day's close and finlab's sim
trades on the next trading day. On a day the exposure changes, the position is the stocks the
base strategy actually holds that day (stopped-out stocks excluded) at their base weights times
the exposure; the rest stays cash. Fees, stop-losses, position limits, execution price and
rebalance dates are the base strategy's own. With exposure fixed at 1 the same machinery
reproduces the base strategy day by day, which the script checks.

"Wins" means a higher daily Sharpe ratio (mean / std x sqrt(252), no risk-free rate) over the
same dates. A filtered strategy passes only if
  1. at least 7 of 9 settings beat the market benchmark (0050 with dividends in Taiwan,
     QQQ with dividends in the US), the original setting beats it in both halves of the period,
     at sim's default cost; US designs must also beat QQQ at 0.1% per side, above T-bills (BIL)
     and without the calendar year of their largest lead;
  2. at least 7 of 9 settings beat the base strategy itself, and the original setting beats the
     base in both halves (US: Sharpe of returns above BIL);
  3. its weekly returns correlate below 0.95 with the base and with every live strategy page.

Run:  pip install finlab TA-Lib      (one base strategy uses RSI)
      python research.py tw            (or us) writes metrics.json and data.csv next to the article:
                                       tw here, us in ../200-day-moving-average-market-filter/
Each base strategy runs in its own process and is cached in the system temp folder. Outside a checkout
of the finlab-website repo the scripts are downloaded from finlab.finance, and check 3 compares with the
base strategy only (the live strategy pages' curves are read from content/strategies-data).
"""
import ast
import contextlib
import io
import json
import os
import runpy
import subprocess
import sys
import tempfile
import urllib.request
from itertools import combinations, product
from pathlib import Path

import numpy as np
import pandas as pd

import finlab
import finlab.backtest
from finlab import config, data

END = "2026-11-06"
OUT = Path(__file__).resolve().parent
# The repo checkout this file sits in (static/blog/<slug>/research.py), if any
REPO = next((p for p in OUT.parents if (p / "content" / "strategies-data").is_dir()), None)
# Each study's tables and weekly curves go next to its article
OUTPUTS = {"tw": OUT, "us": OUT.parent / "200-day-moving-average-market-filter"}
SITE = "https://finlab.finance"
FEE_PER_SIDE = 0.001  # US robustness check

# slug: (download path on finlab.finance, period start = strategy page window start)
BASES = {
    "tw": {
        "30k-stock-selection": ("/blog/30k-stock-selection/validation/strategy.py", "2018-02-12"),
        "ai-agent-quant-trading-tools": ("/blog/ai-agent-quant-trading-tools/validation/strategy.py", "2018-01-31"),
        "ai-prompt-backtest-strategy": ("/blog/ai-prompt-backtest-strategy/validation/strategy.py", "2013-06-28"),
        "ai-quant-research": ("/blog/ai-quant-research/validation/strategy.py", "2015-01-12"),
        "cash-dividend-increase-stock-selection": ("/blog/cash-dividend-increase-stock-selection/strategy.py", "2019-02-11"),
        "convertible-bond-stock-selection": ("/blog/convertible-bond-stock-selection/strategy.py", "2010-03-11"),
        "dupont-analysis-stock-picking-intro": ("/blog/dupont-analysis-stock-picking-intro/validation/strategy.py", "2013-04-24"),
        "etf-0056-replication": ("/blog/etf-0056-replication/validation/strategy.py", "2014-05-30"),
        "f-score-low-pb-revenue": ("/blog/f-score-indicator/low-pb-revenue/strategy.py", "2014-06-30"),
        "financial-stock-strategy": ("/blog/financial-stock-strategy/validation/strategy.py", "2010-01-29"),
        "four-factor-composite": ("/blog/multi-factor-stock-selection-beat-0050/four-factor-composite/strategy.py", "2018-02-12"),
        "high-dividend-etf-comparison": ("/blog/high-dividend-etf-comparison/validation/strategy.py", "2010-01-29"),
        "high-dividend-four-factor": ("/blog/ai-agent-quant-trading-tools/high-dividend-four-factor/strategy.py", "2018-01-31"),
        "institutional-investors-comparison": ("/blog/institutional-investors-comparison/validation/strategy.py", "2015-01-30"),
        "linearfactor": ("/blog/linearfactor/validation/strategy.py", "2010-03-31"),
        "liquidity-risk-analysis-backtest": ("/blog/liquidity-risk-analysis-backtest/validation/strategy.py", "2008-04-30"),
        "monthly-revenue-momentum": ("/blog/monthly-revenue-momentum/validation/strategy.py", "2015-02-11"),
        "monthly-revenue-surprise-strategy": ("/blog/monthly-revenue-surprise-strategy/strategy.py", "2015-02-11"),
        "multi-factor-stock-selection-beat-0050": ("/blog/multi-factor-stock-selection-beat-0050/validation/strategy.py", "2018-02-12"),
        "new-high-strategy": ("/blog/new-high-strategy/validation/strategy.py", "2015-01-12"),
        "operating-margin-three-factor": ("/blog/ai-agent-quant-trading-tools/operating-margin-three-factor/strategy.py", "2018-01-31"),
        "optimize-stock-selection-beating-0050-etf": ("/blog/optimize-stock-selection-beating-0050-etf/validation/strategy.py", "2013-06-28"),
        "pe-pb-peg-backtest": ("/blog/pe-pb-peg-backtest/validation/strategy.py", "2015-03-31"),
        "price-to-sales-ratio-cheap-revenue-stocks": ("/blog/price-to-sales-ratio-cheap-revenue-stocks/validation/strategy.py", "2013-04-30"),
        "python-finlab-optimization": ("/blog/python-finlab-optimization/validation/strategy.py", "2010-06-30"),
        "python-quant-trading-tutorial": ("/blog/python-quant-trading-tutorial/validation/strategy.py", "2007-10-31"),
        "revenue-new-high-stock-selection": ("/blog/revenue-new-high-stock-selection/strategy.py", "2015-02-11"),
        "risk-of-diversification": ("/blog/risk-of-diversification/validation/strategy.py", "2015-02-11"),
        "small-cap-effect": ("/blog/small-cap-effect/validation/strategy.py", "2013-04-30"),
        "strategy-dog-free-cash-flow-stock-screen-backtest": ("/blog/strategy-dog-free-cash-flow-stock-screen-backtest/validation/strategy.py", "2013-04-30"),
        "treasury-stock-repeat-buyback": ("/blog/tw-corporate-actions-event-backtest/treasury-stock-repeat-buyback/strategy.py", "2013-05-02"),
    },
    "us": {
        "us-earnings-surprise-strategy": ("/blog/us-earnings-surprise-strategy/strategy.py", "2016-05-02"),
        "us-quality-value-strategy": ("/blog/us-quality-value-strategy/strategy.py", "2016-05-02"),
    },
}

GRIDS = {
    ("tw", "trend"): {"days": (60, 120, 240), "buffer": (0.0, 0.02, 0.04)},
    ("us", "trend"): {"days": (100, 150, 200), "buffer": (0.0, 0.02, 0.04)},
    ("tw", "vol"): {"lookback": (21, 63, 126), "mult": (0.8, 1.0, 1.2)},
    ("us", "vol"): {"lookback": (21, 63, 126), "mult": (0.8, 1.0, 1.2)},
}

data.truncate_end = END


# ---------------------------------------------------------------------------
# Base strategies: run the page's script unchanged and record its sim() call
# ---------------------------------------------------------------------------
def script_text(path):
    local = REPO / "static" / path.lstrip("/") if REPO else None
    if local and local.exists():
        return local.read_text(encoding="utf-8")
    # the site rejects urllib's default User-Agent
    request = urllib.request.Request(SITE + path, headers={"User-Agent": "finlab-research/1.0"})
    with urllib.request.urlopen(request) as response:
        return response.read().decode("utf-8")


def run_base(market, slug):
    """Run the strategy page's script and return what its last sim() call received and produced."""
    calls = []
    real_sim = finlab.backtest.sim

    def recording_sim(position, *args, **kwargs):
        report = real_sim(position, *args, **kwargs)
        calls.append({"position": position, "args": args, "kwargs": kwargs, "report": report,
                      "market": config.get_market()})
        return report

    finlab.backtest.sim = recording_sim
    argv = sys.argv
    try:
        with tempfile.TemporaryDirectory() as tmp, working_directory(tmp):
            script = Path(tmp) / "strategy.py"
            script.write_text(script_text(BASES[market][slug][0]), encoding="utf-8")
            sys.argv = [str(script)]
            with contextlib.redirect_stdout(io.StringIO()):
                runpy.run_path(str(script), run_name="__main__")
    finally:
        finlab.backtest.sim = real_sim
        sys.argv = argv
    if not calls:
        raise RuntimeError(f"{slug} never called sim()")
    call = calls[-1]
    names = list(sim_signature_names())
    settings = dict(zip(names[1:], call["args"]))
    settings.update(call["kwargs"])
    for key in ("upload", "name", "notification_enable", "line_access_token", "metrics_only"):
        settings.pop(key, None)
    return {"report": call["report"], "position": call["position"], "settings": settings,
            "market": call["market"], "sim_calls": len(calls)}


@contextlib.contextmanager
def working_directory(path):
    """contextlib.chdir for Python 3.10: the strategy scripts write their own files into the cwd."""
    previous = os.getcwd()
    os.chdir(path)
    try:
        yield
    finally:
        os.chdir(previous)


def sim_signature_names():
    import inspect
    return inspect.signature(finlab.backtest.sim).parameters


# ---------------------------------------------------------------------------
# Putting an exposure on top of a base strategy
# ---------------------------------------------------------------------------
def normalize(rows, limit):
    """sim's rule: scale a row down to a total weight of 1 (never up), then cap each stock."""
    rows = rows.fillna(0.0)
    rows = rows.div(rows.abs().sum(axis=1).clip(lower=1), axis=0)
    return rows.clip(-limit, limit) if limit < 1 else rows


def held_on(trades, day):
    """Stocks the base strategy holds after the close of `day` (entered, not yet exited)."""
    alive = (trades["entry_sig_date"] <= day) & (trades["exit_sig_date"].isna() | (trades["exit_sig_date"] > day))
    return set(trades.loc[alive, "stock_id"].astype(str).str.split(" ").str[0])


def trading_days(base):
    """The market's trading calendar (the base's equity curve can start after its first signal)."""
    days = base["market"].get_price("close", adj=False).index
    return days[days <= base["report"].creturn.index[-1]]


def base_weights(base):
    """Weights the base strategy applies at each rebalance, keyed by the signal's trading day.

    sim samples the strategy's position at its rebalance dates and, after a stop-loss, leaves the
    stopped stock out of the next rebalance (stop_trading_next_period) before scaling the row."""
    report = base["report"]
    days = trading_days(base)
    weights = report.position.astype(float)
    # a rebalance due before the next session (e.g. a month-end Saturday) trades on the last data day's signal;
    # later ones are not due yet
    weights = weights[weights.index < days[-1] + pd.offsets.BDay()]
    nominal = weights.index
    # nominal dates such as the 14th of a month may be holidays: sim reads the last trading day before
    weights.index = days[days.searchsorted(nominal, side="right") - 1]
    keep = ~weights.index.duplicated(keep="last")
    weights, nominal = weights[keep], nominal[keep]
    settings = base["settings"]
    stops = any(settings.get(k) is not None for k in ("stop_loss", "take_profit", "trail_stop"))
    if not stops or not settings.get("stop_trading_next_period", True):
        return weights

    trades = report.trades
    stock = trades["stock_id"].astype(str).str.split(" ").str[0]
    limit = base["settings"].get("position_limit", 1)
    raw = pd.DataFrame(base["position"]).astype(float)
    for i in range(1, len(weights)):
        previous, day = weights.index[i - 1], weights.index[i]
        stopped = set(stock[(trades["exit_sig_date"] > previous) & (trades["exit_sig_date"] < day)])
        if not weights.columns.isin(stopped).any():
            continue
        row = raw.loc[:nominal[i]].iloc[[-1]].reindex(columns=weights.columns).fillna(0.0)
        gap = float((normalize(row, limit).iloc[0] - weights.iloc[i]).abs().max())
        if gap > 1e-9:
            raise RuntimeError(f"position at {day.date()} differs from report.position by {gap}")
        row.loc[:, row.columns.isin(stopped)] = 0.0
        weights.iloc[i] = normalize(row, limit).iloc[0].to_numpy()
    return weights


def layered_position(base, exposure):
    """Daily target weights: base weights x exposure; returns (position, rebalance dates)."""
    report = base["report"]
    weights = base_weights(base)
    days = trading_days(base)
    days = days[days >= weights.index[0]]
    exposure = curve_on(exposure, days).fillna(1.0)
    changes = exposure.index[exposure.ne(exposure.shift()) & (exposure.index > days[0])]
    extra = changes.difference(weights.index)

    rows = weights.mul(exposure.reindex(weights.index).fillna(1.0), axis=0)
    if len(extra):
        trades = report.trades
        last = curve_on(weights, extra)
        for day in extra:
            held = held_on(trades, day)
            row = last.loc[day].where(last.columns.isin(held), 0.0)
            rows.loc[day] = row * exposure[day]
        rows = rows.sort_index()
    daily = rows.reindex(days, method="ffill").fillna(0.0)
    return daily, rows.index


def run_layered(base, exposure, **overrides):
    position, rebalance = layered_position(base, exposure)
    settings = dict(base["settings"])
    settings.pop("resample_offset", None)
    if settings.get("resample") is not None:
        settings["resample"] = pd.DatetimeIndex(rebalance)
    settings.update(overrides)
    report = finlab.backtest.sim(position, upload=False, market=base["market"], **settings)
    # sim drops the flat start of an equity curve; days in cash before the first purchase stay at 1
    return report.creturn.reindex(position.index).ffill().fillna(1.0)


def trend_exposure(market, days, buffer):
    index = market_index(market)
    average = index.rolling(days).mean()
    state = pd.Series(np.nan, index=index.index)
    state[index >= average] = 1.0
    state[index < average * (1 - buffer)] = 0.0
    return state.ffill().fillna(1.0)


def vol_exposure(base, lookback, mult):
    returns = base["report"].creturn.pct_change().dropna()
    sigma = returns.rolling(lookback).std() * 252 ** 0.5
    target = sigma.expanding().median() * mult
    raw = (target / sigma).clip(upper=1.0)
    month_ends = returns.groupby(returns.index.to_period("M")).tail(1).index
    return raw.reindex(month_ends).reindex(returns.index).ffill().fillna(1.0)


def exposure_for(market, design, base, setting):
    if design == "trend":
        return trend_exposure(market, *setting)
    return vol_exposure(base, *setting)


# ---------------------------------------------------------------------------
# Benchmarks and statistics
# ---------------------------------------------------------------------------
_cache = {}


def cached(key, load):
    if key not in _cache:
        _cache[key] = load()
    return _cache[key]


def market_index(market):
    if market == "tw":
        return cached("taiex", lambda: data.get("benchmark_return:發行量加權股價報酬指數").iloc[:, 0].dropna())
    return us_fund("QQQ")


def us_fund(ticker):
    def load():
        with data.market("us_fund"):
            return data.get("us_fund_price:adj_close")
    return cached("us_fund", load)[ticker].dropna()


def benchmark(market):
    if market == "tw":
        return cached("0050", lambda: data.get("etl:adj_close")["0050"].dropna())
    return us_fund("QQQ")


def daily_sharpe(returns):
    returns = returns.dropna()
    return float(returns.mean() / returns.std() * 252 ** 0.5)


def curve_on(series, index):
    return series.reindex(series.index.union(index)).ffill().reindex(index)


def excess_sharpe(curve):
    bil = curve_on(us_fund("BIL"), curve.index).pct_change()
    return daily_sharpe(curve.pct_change() - bil)


def score(market, curve):
    """The Sharpe used against the base: plain in Taiwan, above T-bills in the US."""
    return daily_sharpe(curve.pct_change()) if market == "tw" else excess_sharpe(curve)


def window(curve, start):
    curve = curve.loc[start:END].dropna()
    return curve / curve.iloc[0]


def stats(curve):
    years = (curve.index[-1] - curve.index[0]).days / 365.25
    return {
        "start": str(curve.index[0].date()), "end": str(curve.index[-1].date()),
        "cagr": round(float((curve.iloc[-1] / curve.iloc[0]) ** (1 / years) - 1), 6),
        "sharpe": round(daily_sharpe(curve.pct_change()), 6),
        "mdd": round(float((curve / curve.cummax() - 1).min()), 6),
        "volatility": round(float(curve.pct_change().std() * 252 ** 0.5), 6),
    }


def halves(curve):
    middle = curve.index[len(curve) // 2]
    return curve.loc[:middle], curve.loc[middle:]


def yearly(curve):
    ends = curve.resample("YE").last()
    return ends / ends.shift(1, fill_value=curve.iloc[0]) - 1


def without_best_year(curve, bench):
    lead = yearly(curve) - yearly(bench)
    best = lead.idxmax().year
    keep = curve.index.year != best
    return {"year": int(best), "lead": round(float(lead.max()), 6),
            "sharpe": round(daily_sharpe(curve.pct_change()[keep]), 6),
            "bench_sharpe": round(daily_sharpe(bench.pct_change()[keep]), 6)}


def weekly_returns(curve):
    return curve.resample("W-FRI").last().pct_change().dropna()


def correlation(a, b):
    a, b = a.align(b, join="inner")
    return (round(float(a.corr(b)), 6), len(a)) if len(a) >= 60 else (None, len(a))


def page_correlation(curve, page):
    """Correlation of returns between the dates a strategy page samples (about every 5 trading days)."""
    dates = page.index[(page.index >= curve.index[0]) & (page.index <= curve.index[-1])]
    return correlation(curve_on(curve, dates).pct_change().dropna(), page.reindex(dates).pct_change().dropna())


def live_strategies():
    """Equity curves of the strategy pages live before this study (repo checkout only)."""
    out = {}
    if REPO is None:
        return out
    for path in sorted((REPO / "content" / "strategies-data").glob("*.json")):
        page = json.loads(path.read_text(encoding="utf-8"))
        if page.get("downloadPy", "").startswith(f"/blog/{OUT.name}/"):
            continue  # pages opened by this study are its own results, not earlier strategies
        series = page.get("equitySeries")
        if isinstance(series, str):
            series = ast.literal_eval(series)
        if not series:
            continue
        curve = pd.Series({pd.Timestamp(p["date"]): p["value"] for p in series}).sort_index()
        out[page["slug"]] = curve
    return out


# ---------------------------------------------------------------------------
# Study
# ---------------------------------------------------------------------------
def study_base(market, slug, live):
    start = BASES[market][slug][1]
    # read the benchmarks first: a base script may narrow finlab's stock universe (data.set_universe)
    benchmark(market), market_index(market)
    if market == "us":
        us_fund("BIL")
    base = run_base(market, slug)
    base_curve = window(base["report"].creturn, start)
    bench = window(curve_on(benchmark(market), base_curve.index), start)
    base_stats = stats(base_curve)
    base_stats["score"] = round(score(market, base_curve), 6)

    # The same machinery with exposure fixed at 1 must give back the base strategy.
    ones = pd.Series(1.0, index=base["report"].creturn.index)
    replay = window(run_layered(base, ones), start)
    replay_gap = float((replay / base_curve.reindex(replay.index) - 1).abs().max())

    base_first, base_second = halves(base_curve)
    result = {
        "slug": slug, "market": market, "sim_calls": base["sim_calls"],
        "settings": {k: (str(v) if not isinstance(v, (int, float, str, bool, type(None))) else v)
                     for k, v in base["settings"].items() if k != "position"},
        "base": base_stats, "replay_gap": round(replay_gap, 8),
        "benchmark": stats(bench),
        "base_halves": [round(score(market, part), 6) for part in (base_first, base_second)],
        "designs": {},
    }
    curves = {"base": base_curve, "benchmark": bench}
    for design in ("trend", "vol"):
        grid = GRIDS[(market, design)]
        original = tuple(values[1] for values in grid.values())
        cells, layered = {}, {}
        for setting in product(*grid.values()):
            exposure = exposure_for(market, design, base, setting)
            curve = window(run_layered(base, exposure), start)
            layered[setting] = curve
            in_window = exposure.reindex(curve.index, method="ffill").fillna(1.0)
            cell = stats(curve)
            cell["score"] = round(score(market, curve), 6)
            cell["exposure"] = round(float(in_window.mean()), 6)
            cell["beats_benchmark"] = cell["sharpe"] > result["benchmark"]["sharpe"]
            cell["beats_base"] = cell["score"] > base_stats["score"]
            cells[str(setting)] = cell
        curve = layered[original]
        bench_on = bench.reindex(curve.index)
        first, second = halves(curve)
        b_first, b_second = halves(bench_on)
        orig = cells[str(original)]
        design_result = {
            "original": str(original), "grid": cells,
            "benchmark_wins": sum(c["beats_benchmark"] for c in cells.values()),
            "base_wins": sum(c["beats_base"] for c in cells.values()),
            "halves": [round(daily_sharpe(p.pct_change()), 6) for p in (first, second)],
            "benchmark_halves": [round(daily_sharpe(p.pct_change()), 6) for p in (b_first, b_second)],
            "score_halves": [round(score(market, p), 6) for p in (first, second)],
            "sharpe_change": round(orig["sharpe"] - base_stats["sharpe"], 6),
            "mdd_change": round(orig["mdd"] - base_stats["mdd"], 6),
            "corr_base": correlation(weekly_returns(curve), weekly_returns(base_curve))[0],
        }
        checks = {
            "grid_vs_benchmark": design_result["benchmark_wins"] >= 7,
            "halves_vs_benchmark": all(s > b for s, b in zip(design_result["halves"], design_result["benchmark_halves"])),
            "cost_vs_benchmark": orig["beats_benchmark"],
            "grid_vs_base": design_result["base_wins"] >= 7,
            "halves_vs_base": all(s > b for s, b in zip(design_result["score_halves"], result["base_halves"])),
        }
        if market == "us":
            exposure = exposure_for(market, design, base, original)
            costly = window(run_layered(base, exposure, fee_ratio=FEE_PER_SIDE, tax_ratio=0), start)
            design_result["with_cost"] = stats(costly)
            design_result["excess_sharpe"] = round(excess_sharpe(curve), 6)
            design_result["benchmark_excess_sharpe"] = round(excess_sharpe(bench_on), 6)
            design_result["without_best_year"] = without_best_year(curve, bench_on)
            checks["cost_vs_benchmark"] = orig["beats_benchmark"] and design_result["with_cost"]["sharpe"] > result["benchmark"]["sharpe"]
            checks["above_tbills"] = design_result["excess_sharpe"] > design_result["benchmark_excess_sharpe"]
            wby = design_result["without_best_year"]
            checks["without_best_year"] = wby["sharpe"] > wby["bench_sharpe"]
        corrs = {other: page_correlation(curve, page) for other, page in live.items()}
        design_result["corr_live_max"] = max(((c, n, o) for o, (c, n) in corrs.items() if c is not None), default=None)
        checks["distinct"] = (design_result["corr_base"] is not None and design_result["corr_base"] < 0.95
                              and all(c is None or c < 0.95 for c, _ in corrs.values()))
        design_result["checks"] = {k: bool(v) for k, v in checks.items()}
        design_result["passes"] = all(design_result["checks"].values())
        result["designs"][design] = design_result
        curves[design] = curve
    return result, curves


def off_days(market, curve, trade_at_price="close"):
    """How a curve did on the days the original trend filter held cash, and on the other days.

    A signal at close T is traded at the next day's close (or open), so the return ending T+1 still
    belongs to the old position when trading at the close: the first day in cash is T+2."""
    grid = GRIDS[(market, "trend")]
    exposure = trend_exposure(market, grid["days"][1], grid["buffer"][1])
    held = curve_on(exposure, curve.index).shift(2 if trade_at_price == "close" else 1)
    returns = curve.pct_change()
    out = {"exits": int(((held == 0) & (held.shift() == 1)).sum())}
    for label, mask in (("out", held == 0), ("in", held == 1)):
        part = returns[mask].dropna()
        out[label] = {"share": round(len(part) / returns.count(), 6),
                      "annual_return": round(float(part.mean() * 252), 6),
                      "volatility": round(float(part.std() * 252 ** 0.5), 6),
                      "sharpe": round(daily_sharpe(part), 6)}
    return out


def assemble(market, cache):
    """metrics.json (the numbers behind the article) and data.csv (weekly curves) for one market."""
    finlab.login()
    results, weekly = {}, {}
    for slug in BASES[market]:
        results[slug] = json.loads((cache / f"{slug}.json").read_text(encoding="utf-8"))
        curves = pd.read_csv(cache / f"{slug}.csv", index_col=0, parse_dates=True)
        trade_at = results[slug]["settings"].get("trade_at_price", "close")
        results[slug]["off_days"] = off_days(market, curves["base"].dropna(), trade_at)
        for column in ("base", "trend", "vol"):
            weekly[f"{slug}:{column}"] = curves[column].dropna().resample("W-FRI").last()
    first = min(BASES[market][slug][1] for slug in BASES[market])
    bench = benchmark(market).loc[first:END]
    weekly[f"{'0050' if market == 'tw' else 'QQQ'} total return"] = (bench / bench.iloc[0]).resample("W-FRI").last()
    passing = [(slug, design) for slug, r in results.items() for design, d in r["designs"].items() if d["passes"]]
    pairs = {}
    for (a, da), (b, db) in combinations(passing, 2):
        ra = weekly[f"{a}:{da}"].pct_change().dropna()
        rb = weekly[f"{b}:{db}"].pct_change().dropna()
        pairs[f"{a}:{da} x {b}:{db}"] = correlation(ra, rb)[0]
    metrics = {
        "end": END, "grids": {d: GRIDS[(market, d)] for d in ("trend", "vol")},
        "index_off_days": {"start": first, **off_days(market, market_index(market).loc[first:END])},
        "index_exits": {f"{days}/{buffer}": int(((e == 0) & (e.shift() == 1)).loc[first:END].sum())
                        for days, buffer in product(*GRIDS[(market, "trend")].values())
                        for e in [trend_exposure(market, days, buffer)]},
        "passing": [f"{slug}:{design}" for slug, design in passing], "passing_correlations": pairs,
        "strategies": results, "run_date": str(pd.Timestamp.today().date()),
        "finlab_version": finlab.__version__, "pandas_version": pd.__version__,
    }
    out = OUTPUTS[market]
    out.mkdir(exist_ok=True)
    (out / "metrics.json").write_text(json.dumps(metrics, ensure_ascii=False, indent=1, default=str),
                                                encoding="utf-8")
    frame = pd.DataFrame(weekly)
    frame.index.name = "date"
    frame.round(6).to_csv(out / "data.csv")


def main():
    market = sys.argv[1] if len(sys.argv) > 1 else "tw"
    only = sys.argv[2:] or list(BASES[market])
    cache = Path(tempfile.gettempdir()) / f"finlab-market-filter-{market}"
    cache.mkdir(exist_ok=True)
    if len(only) > 1:
        # One process per strategy: some scripts change finlab's global state (e.g. data.set_universe).
        for slug in only:
            if not (cache / f"{slug}.json").exists():
                subprocess.run([sys.executable, __file__, market, slug], check=False)
        if all((cache / f"{slug}.json").exists() for slug in BASES[market]):
            assemble(market, cache)
        return
    finlab.login()
    live = live_strategies()
    for slug in only:
        target = cache / f"{slug}.json"
        if target.exists():
            continue
        result, curves = study_base(market, slug, live)
        target.write_text(json.dumps(result, ensure_ascii=False, indent=1, default=str), encoding="utf-8")
        pd.DataFrame(curves).round(6).to_csv(cache / f"{slug}.csv", index_label="date")
        print(slug, json.dumps({d: (r["passes"], r["benchmark_wins"], r["base_wins"], r["sharpe_change"])
                                for d, r in result["designs"].items()}), "replay", result["replay_gap"], flush=True)


if __name__ == "__main__":
    main()