Portfolio_Risk_Management/quantcore/cli.py

"""Command-line interface: ``quantcore <command>``.

    quantcore setup      [--offline] [--max-per-region N]   build the universe once (candidates,
                                                             metadata, full price + FX history)
    quantcore reselect                                       re-run only the selection step
    quantcore clean                                          re-apply price cleaning to stored data
    quantcore update     [--full]                            fetch the newest prices / FX
    quantcore snapshot   [--name NAME]                       freeze the data for reproducible demos
    quantcore status                                         what the store holds, cuOpt health
    quantcore risk       SYMBOL [SYMBOL ...] [--horizon week] [--freq daily] [--window 200] [--lam 0.94]
    quantcore performance SYMBOL [SYMBOL ...] [--freq weekly]
    quantcore optimize   [--cash 1e6] [--cvar-limit 50000] [--confidence 0.95] [--horizon week] ...
    quantcore serve      [--host 127.0.0.1] [--port 5001]    run the HTTP API (+ the test bench at /ui/)

These commands run the services in this process and need the data directory locally. To drive a
*running* service over HTTP instead -- from any machine, without the data -- use ``quantcore-api``
(:mod:`quantcore.apicli`), whose ``selftest`` also checks a deployment endpoint by endpoint.

The commands that write the data (setup, reselect, update, clean, snapshot) take the store's write
lock, so they never run at the same time as each other or as a data job of the API; one that finds
the lock taken exits with status 3 (on Kubernetes the CronJob retries it).
"""

from __future__ import annotations

import argparse
import json
import logging
import sys
from pathlib import Path

import pandas as pd

from quantcore.config import Settings, get_settings

log = logging.getLogger("quantcore.cli")


def _settings(args) -> Settings:
    s = get_settings()
    if getattr(args, "data_dir", None):
        s = s.model_copy(update={"data_dir": args.data_dir})
    if getattr(args, "cuopt_url", None):
        s = s.model_copy(update={"cuopt_url": args.cuopt_url})
    return s


def _write_lock(s: Settings, command: str):
    from quantcore.data.store import DataStore

    return DataStore(s).write_lock(f"quantcore {command}")


def cmd_setup(args) -> int:
    from quantcore.universe.setup_service import SetupService

    s = _settings(args)
    with _write_lock(s, "setup"):
        report = SetupService(s).run(allow_web=not args.offline, max_per_region=args.max_per_region)
    print(json.dumps(report.as_dict(), indent=2, default=str))
    return 0


def cmd_reselect(args) -> int:
    from quantcore.universe.setup_service import SetupService

    s = _settings(args)
    with _write_lock(s, "reselect"):
        report = SetupService(s).rebuild_selection()
    print(json.dumps(report.as_dict(), indent=2, default=str))
    return 0


def cmd_update(args) -> int:
    from quantcore.data.download_service import DataDownloadService

    s = _settings(args)
    with _write_lock(s, "update"):
        report = DataDownloadService(s).update(full=args.full)
    print(json.dumps(report.as_dict(), indent=2, default=str))
    return 0


def cmd_clean(args) -> int:
    from quantcore.data.download_service import DataDownloadService

    s = _settings(args)
    with _write_lock(s, "clean"):
        report = DataDownloadService(s).clean()
    print(json.dumps(report.as_dict(), indent=2, default=str))
    return 0


def _rounded(table: pd.DataFrame, digits: int) -> pd.DataFrame:
    """Round the numeric columns only: pandas warns when asked to round the date columns too."""
    out = table.copy()
    numeric = out.select_dtypes("number").columns
    out[numeric] = out[numeric].round(digits)
    return out


def cmd_snapshot(args) -> int:
    from quantcore.data.store import DataStore

    s = _settings(args)
    with _write_lock(s, "snapshot"):
        path = DataStore(s).create_snapshot(args.name)
    print(path)
    return 0


def cmd_status(args) -> int:
    from quantcore.data.store import DataStore
    from quantcore.optimization.solvers import CuOptSolver

    s = _settings(args)
    out = {"store": DataStore(s).status(), "cuopt": CuOptSolver(s).health(), "cuopt_url": s.cuopt_url}
    print(json.dumps(out, indent=2, default=str))
    return 0


def cmd_risk(args) -> int:
    from quantcore.data.market_data import MarketDataService
    from quantcore.measurement.risk import RiskService

    md = MarketDataService(_settings(args), snapshot=args.snapshot)
    table = RiskService(md).asset_risk(args.symbols, freq=args.freq, window=args.window, lam=args.lam,
                                       horizon=args.horizon, currency=args.currency)
    with pd.option_context("display.width", 200, "display.max_columns", 30):
        print(_rounded(table, 5).T)
    return 0


def cmd_performance(args) -> int:
    from quantcore.data.market_data import MarketDataService
    from quantcore.measurement.performance import PerformanceService

    s = _settings(args)
    md = MarketDataService(s, snapshot=args.snapshot)
    table = PerformanceService(md, s.risk_free_rate).asset_performance(args.symbols, freq=args.freq,
                                                                       currency=args.currency, start=args.start)
    with pd.option_context("display.width", 200, "display.max_columns", 30):
        print(_rounded(table, 4).T)
    return 0


def cmd_optimize(args) -> int:
    from quantcore.data.market_data import MarketDataService
    from quantcore.optimization.service import OptimizationRequest, OptimizationService

    s = _settings(args)
    md = MarketDataService(s, snapshot=args.snapshot)
    kwargs = {}
    if args.request:
        kwargs = json.loads(open(args.request).read())
    kwargs.setdefault("cash", args.cash)
    kwargs.setdefault("horizon", args.horizon)
    kwargs.setdefault("confidence", args.confidence)
    if args.cvar_limit is not None:
        kwargs["cvar_limit"] = args.cvar_limit
    if args.symbols:
        kwargs["symbols"] = args.symbols
    kwargs.setdefault("max_assets", args.max_assets)
    kwargs.setdefault("scenario_model", args.scenario_model)
    if args.n_scenarios is not None:
        kwargs.setdefault("n_scenarios", args.n_scenarios)
    kwargs.setdefault("sampling", args.sampling)
    kwargs.setdefault("importance_sampling", args.importance_sampling)
    kwargs.setdefault("row_generation", args.row_generation)
    kwargs.setdefault("strategy", args.strategy)
    kwargs.setdefault("solver", args.solver)
    if args.holdings:
        kwargs["holdings"] = json.loads(args.holdings)
        kwargs["mode"] = "rebalance"
    req = OptimizationRequest(**kwargs)
    res = OptimizationService(s, md).optimize(req)
    if args.json:
        print(json.dumps(res.as_dict(), indent=2, default=str))
    else:
        print(f"status={res.status} solver={res.solver} reproducible={res.reproducibility} scenarios={res.scenario_hash} "
              f"key={res.scenario_key['id'] if res.scenario_key else None} "
              f"wealth={res.wealth:,.0f} invested={res.invested:,.0f} cash_end={res.cash_end:,.0f} cost={res.total_cost:,.0f}")
        print(f"E[P&L]={res.expected_pnl:,.0f} ({res.expected_return:.3%}) VaR={res.var:,.0f} CVaR={res.cvar:,.0f} "
              f"limit={res.cvar_limit} @ {res.confidence:.0%} over {res.horizon} ({res.horizon_periods} periods)")
        if res.positions:
            df = pd.DataFrame(res.positions)[["symbol", "name", "sector", "price", "shares", "trade", "value", "weight"]]
            with pd.option_context("display.width", 200, "display.max_rows", 500):
                print(df.round(3).to_string(index=False))
        print("sectors:", {k: round(v, 3) for k, v in res.sector_exposure.items()})
        print("timings:", res.timings, "warnings:", res.warnings)
    return 0 if res.status in ("Optimal", "FeasibleFound") else 2


def cmd_serve(args) -> int:
    import uvicorn

    s = _settings(args)
    uvicorn.run("quantcore.api.app:app", host=args.host or s.api_host, port=args.port or s.api_port, reload=False)
    return 0


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(prog="quantcore", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("--data-dir", type=Path, help="override QUANTCORE_DATA_DIR")
    p.add_argument("--cuopt-url", help="override QUANTCORE_CUOPT_URL")
    sub = p.add_subparsers(dest="command", required=True)

    sp = sub.add_parser("setup", help="build the universe and download all history (run once)")
    sp.add_argument("--offline", action="store_true", help="use the static candidate lists shipped with the package")
    sp.add_argument("--max-per-region", type=int, default=None, help="limit candidates per region (testing)")
    sp.set_defaults(func=cmd_setup)

    sub.add_parser("reselect", help="re-run only the universe selection on stored data").set_defaults(func=cmd_reselect)

    up = sub.add_parser("update", help="fetch the newest prices and FX rates")
    up.add_argument("--full", action="store_true", help="re-download the complete history")
    up.set_defaults(func=cmd_update)

    sub.add_parser("clean", help="re-apply the price cleaning to the stored history").set_defaults(func=cmd_clean)

    sn = sub.add_parser("snapshot", help="freeze universe/prices/fx under snapshots/<name>")
    sn.add_argument("--name", default=None)
    sn.set_defaults(func=cmd_snapshot)

    sub.add_parser("status", help="store status and cuOpt health").set_defaults(func=cmd_status)

    rk = sub.add_parser("risk", help="per-asset volatility, VaR and CVaR")
    rk.add_argument("symbols", nargs="+")
    rk.add_argument("--freq", default="daily", choices=["daily", "weekly", "monthly"])
    rk.add_argument("--horizon", default="day")
    rk.add_argument("--window", type=int, default=200)
    rk.add_argument("--lam", type=float, default=0.94)
    rk.add_argument("--currency", default="base", choices=["base", "local"])
    rk.add_argument("--snapshot", default=None)
    rk.set_defaults(func=cmd_risk)

    pf = sub.add_parser("performance", help="per-asset performance statistics")
    pf.add_argument("symbols", nargs="+")
    pf.add_argument("--freq", default="daily", choices=["daily", "weekly", "monthly"])
    pf.add_argument("--currency", default="base", choices=["base", "local"])
    pf.add_argument("--start", default=None)
    pf.add_argument("--snapshot", default=None)
    pf.set_defaults(func=cmd_performance)

    op = sub.add_parser("optimize", help="mean-CVaR MILP optimisation")
    op.add_argument("--request", help="JSON file with an OptimizationRequest (fields override the flags)")
    op.add_argument("--cash", type=float, default=1_000_000.0)
    op.add_argument("--holdings", help='JSON dict {"NESN.SW": 100, ...} -> rebalancing mode')
    op.add_argument("--symbols", nargs="*", default=None)
    op.add_argument("--max-assets", type=int, default=100)
    op.add_argument("--horizon", default="week")
    op.add_argument("--confidence", type=float, default=0.95)
    op.add_argument("--cvar-limit", type=float, default=None, help="max CVaR in CHF over the horizon")
    op.add_argument("--scenario-model", default="empirical", choices=["gbm", "empirical", "gaussian_exact"])
    op.add_argument("--n-scenarios", type=int, default=None,
                    help="default: 16 per candidate within [4096, 16384]; rounded up to a power of two")
    op.add_argument("--sampling", default="sobol", choices=["sobol", "mc", "antithetic", "lhs", "balanced"])
    from quantcore.optimization.service import IMPORTANCE_SAMPLING_MODES  # lazy, like every command's

    op.add_argument("--importance-sampling", default="auto",
                    choices=list(IMPORTANCE_SAMPLING_MODES),
                    help="'market': tilt towards the tail of the cap-weighted portfolio; "
                         "'adaptive': towards a first-pass solution")
    op.add_argument("--row-generation", default="auto", choices=["auto", "on", "off"])
    op.add_argument("--strategy", default="auto", choices=["auto", "milp", "large_scale"],
                    help="large_scale: relaxation, tail rows, screening, round-and-repair (large universes)")
    op.add_argument("--solver", default="auto", choices=["auto", "cuopt", "highs"])
    op.add_argument("--snapshot", default=None)
    op.add_argument("--json", action="store_true", help="print the full result as JSON")
    op.set_defaults(func=cmd_optimize)

    sv = sub.add_parser("serve", help="run the HTTP API and the browser test bench (/ui/)")
    sv.add_argument("--host", default=None)
    sv.add_argument("--port", type=int, default=None)
    sv.set_defaults(func=cmd_serve)
    return p


def main(argv: list[str] | None = None) -> int:
    from quantcore.data.store import StoreBusyError

    args = build_parser().parse_args(argv)
    try:
        return args.func(args)
    except StoreBusyError as exc:
        print(f"error: {exc}.\n       Wait for it to finish (quantcore-api jobs lists the API's jobs).", file=sys.stderr)
        return 3
    except FileNotFoundError as exc:
        if "run the setup service first" not in str(exc):
            raise
        # no traceback for the commonest first-day mistake: a command before the data exists
        print(f"error: no market data yet ({str(exc).split(';')[0]}).\n"
              "       Build it first: quantcore setup   (on a DGX Spark: qc setup --offline),\n"
              "       or import a prepared data set: qc data-import FILE", file=sys.stderr)
        return 2


if __name__ == "__main__":
    sys.exit(main())