Add JSON output for market data commands

README.md

@@ -35,10 +35,21 @@ young zt-pool           # limit-up (涨停) / limit-down / 炸板 pool
 young flow              # north-bound + main-capital fund flow
 young a -d 20260530     # historical date (YYYYMMDD)
 young a --refresh       # bypass cache, force re-fetch
+young indices --json    # pipe-friendly JSON for scripts
+young a --zt --json     # JSON for limit-up / limit-down / broken-board pools
 young update            # upgrade young-stock-cli in the current Python env
 young --help
 ```
 
+JSON output is available on the market data commands, so pipeline users can
+skip the rich tables:
+
+```bash
+young indices --json | jq '.indices[] | .f12'
+young flow --json | jq '.flow'
+young global --json | jq '.hk[] | {symbol, price}'
+```
+
 ### Example output (`young indices`)
 
 ```

src/young_stock/cli.py

@@ -1,8 +1,11 @@
 """young-stock-cli command line interface."""
 from __future__ import annotations
 
+import dataclasses
+import json
 import subprocess
 import sys
+from typing import Any
 
 import click
 
@@ -18,11 +21,36 @@ def cli() -> None:
     pass
 
 
-def _run(market: str, date: str | None, refresh: bool) -> None:
+def _json_default(value: Any) -> Any:
+    if dataclasses.is_dataclass(value):
+        return dataclasses.asdict(value)
+    if hasattr(value, "to_dict"):
+        return value.to_dict()
+    raise TypeError(f"Object of type {type(value).__name__} is not JSON serializable")
+
+
+def _echo_json(payload: Any) -> None:
+    click.echo(json.dumps(payload, ensure_ascii=False, default=_json_default))
+
+
+def _run(market: str, date: str | None, refresh: bool, as_json: bool = False) -> None:
     if refresh:
         _core.NO_CACHE = True
     _core.cache_clear_old(days=7)
     date_str = date or _core.nearest_trade_date()
+    if as_json:
+        if market == "a":
+            _echo_json(_a_share_payload(date_str))
+        elif market == "hk":
+            _echo_json(_hk_market_payload(date_str))
+        elif market == "us":
+            _echo_json(_us_market_payload(date_str))
+        elif market == "global":
+            _echo_json(_global_market_payload(date_str))
+        else:
+            click.echo(f"unknown market: {market}", err=True)
+            sys.exit(1)
+        return
     if market == "a":
         _core.run_a_share(date_str)
     elif market == "hk":
@@ -38,34 +66,92 @@ def _run(market: str, date: str | None, refresh: bool) -> None:
 
 _date_opt = click.option("--date", "-d", default=None, help="Trade date YYYYMMDD (default: nearest trade day).")
 _refresh_opt = click.option("--refresh", is_flag=True, help="Skip cache and force re-fetch.")
+_json_opt = click.option("--json", "as_json", is_flag=True, help="Output raw data as JSON.")
+
+
+def _a_share_payload(date_str: str) -> dict[str, Any]:
+    zt = _core.get_zt_pool(date_str)
+    dt = _core.get_dt_pool(date_str)
+    zb = _core.get_zb_pool(date_str)
+    return {
+        "date": date_str,
+        "indices": _core.get_index(date_str),
+        "zt_pool": zt,
+        "dt_pool": dt,
+        "zb_pool": zb,
+        "flow": _core.get_fund_flow(date_str),
+    }
+
+
+def _zt_payload(date_str: str) -> dict[str, Any]:
+    return {
+        "date": date_str,
+        "zt_pool": _core.get_zt_pool(date_str),
+        "dt_pool": _core.get_dt_pool(date_str),
+        "zb_pool": _core.get_zb_pool(date_str),
+    }
+
+
+def _us_market_payload(date_str: str) -> dict[str, Any]:
+    symbols = {"^GSPC": "标普 500", "^IXIC": "纳斯达克"}
+    return {"date": date_str, "indices": _core.fetch_us_indices_sina(symbols, date_str)}
+
+
+def _hk_market_payload(date_str: str) -> dict[str, Any]:
+    symbols = {"^HSI": "恒生指数", "^HSCE": "国企指数", "HSTECH.HK": "恒生科技指数"}
+    return {"date": date_str, "indices": _core.fetch_hk_indices_tencent(symbols, date_str)}
+
+
+def _global_market_payload(date_str: str) -> dict[str, Any]:
+    return {
+        "date": date_str,
+        "a": _core.get_index(date_str),
+        "hk": _hk_market_payload(date_str)["indices"],
+        "us": _us_market_payload(date_str)["indices"],
+    }
 
 
 @cli.command(help="A-share after-hours dashboard: indices, ZT/DT pool, fund flow, boards.")
 @_date_opt
 @_refresh_opt
-def a(date: str | None, refresh: bool) -> None:
-    _run("a", date, refresh)
+@_json_opt
+@click.option("--zt", is_flag=True, help="Only show limit-up/down pool data.")
+def a(date: str | None, refresh: bool, as_json: bool, zt: bool) -> None:
+    if refresh:
+        _core.NO_CACHE = True
+    date_str = date or _core.nearest_trade_date()
+    if zt:
+        payload = _zt_payload(date_str)
+        if as_json:
+            _echo_json(payload)
+        else:
+            _core.print_zt_analysis(payload["zt_pool"], payload["dt_pool"], payload["zb_pool"])
+        return
+    _run("a", date, refresh, as_json=as_json)
 
 
 @cli.command(help="Hong Kong market after-hours snapshot.")
 @_date_opt
 @_refresh_opt
-def hk(date: str | None, refresh: bool) -> None:
-    _run("hk", date, refresh)
+@_json_opt
+def hk(date: str | None, refresh: bool, as_json: bool) -> None:
+    _run("hk", date, refresh, as_json=as_json)
 
 
 @cli.command(help="US market after-hours snapshot.")
 @_date_opt
 @_refresh_opt
-def us(date: str | None, refresh: bool) -> None:
-    _run("us", date, refresh)
+@_json_opt
+def us(date: str | None, refresh: bool, as_json: bool) -> None:
+    _run("us", date, refresh, as_json=as_json)
 
 
 @cli.command(name="global", help="Global indices snapshot (A + HK + US).")
 @_date_opt
 @_refresh_opt
-def global_(date: str | None, refresh: bool) -> None:
-    _run("global", date, refresh)
+@_json_opt
+def global_(date: str | None, refresh: bool, as_json: bool) -> None:
+    _run("global", date, refresh, as_json=as_json)
 
 
 @cli.command(help="Update young-stock-cli with the current Python environment.")
@@ -87,35 +173,45 @@ def update(pre: bool, user_install: bool) -> None:
 @cli.command(help="Show A-share major indices only.")
 @_date_opt
 @_refresh_opt
-def indices(date: str | None, refresh: bool) -> None:
+@_json_opt
+def indices(date: str | None, refresh: bool, as_json: bool) -> None:
     if refresh:
         _core.NO_CACHE = True
     date_str = date or _core.nearest_trade_date()
     data = _core.get_index(date_str)
+    if as_json:
+        _echo_json({"date": date_str, "indices": data})
+        return
     _core.print_index(data)
 
 
 @cli.command(name="zt-pool", help="Show A-share limit-up (涨停) pool.")
 @_date_opt
 @_refresh_opt
-def zt_pool(date: str | None, refresh: bool) -> None:
+@_json_opt
+def zt_pool(date: str | None, refresh: bool, as_json: bool) -> None:
     if refresh:
         _core.NO_CACHE = True
     date_str = date or _core.nearest_trade_date()
-    zt = _core.get_zt_pool(date_str)
-    dt = _core.get_dt_pool(date_str)
-    zb = _core.get_zb_pool(date_str)
-    _core.print_zt_analysis(zt, dt, zb)
+    payload = _zt_payload(date_str)
+    if as_json:
+        _echo_json(payload)
+        return
+    _core.print_zt_analysis(payload["zt_pool"], payload["dt_pool"], payload["zb_pool"])
 
 
 @cli.command(help="Show A-share fund flow (north-bound, main capital).")
 @_date_opt
 @_refresh_opt
-def flow(date: str | None, refresh: bool) -> None:
+@_json_opt
+def flow(date: str | None, refresh: bool, as_json: bool) -> None:
     if refresh:
         _core.NO_CACHE = True
     date_str = date or _core.nearest_trade_date()
     flow_data = _core.get_fund_flow(date_str)
+    if as_json:
+        _echo_json({"date": date_str, "flow": flow_data})
+        return
     _core.print_fund_flow(flow_data)
 
 

tests/test_cli.py

@@ -1,3 +1,4 @@
+import json
 import sys
 from types import SimpleNamespace
 
@@ -35,6 +36,82 @@ def test_cli_subcommands_registered():
         assert sub in result.output, f"subcommand `{sub}` missing from help"
 
 
+def test_indices_json_outputs_parseable_payload(monkeypatch):
+    from click.testing import CliRunner
+
+    monkeypatch.setattr(cli_module._core, "nearest_trade_date", lambda: "20260529")
+    monkeypatch.setattr(
+        cli_module._core,
+        "get_index",
+        lambda date: [{"f12": "000001", "f14": "上证指数", "f2": 3000.12}],
+    )
+
+    result = CliRunner().invoke(cli, ["indices", "--json"])
+
+    assert result.exit_code == 0
+    payload = json.loads(result.output)
+    assert payload["date"] == "20260529"
+    assert payload["indices"][0]["f12"] == "000001"
+
+
+def test_a_zt_json_outputs_pool_payload(monkeypatch):
+    from click.testing import CliRunner
+
+    monkeypatch.setattr(cli_module._core, "nearest_trade_date", lambda: "20260529")
+    monkeypatch.setattr(cli_module._core, "get_zt_pool", lambda date: {"data": {"tc": 2}})
+    monkeypatch.setattr(cli_module._core, "get_dt_pool", lambda date: {"data": {"tc": 1}})
+    monkeypatch.setattr(cli_module._core, "get_zb_pool", lambda date: {"data": {"tc": 3}})
+
+    result = CliRunner().invoke(cli, ["a", "--zt", "--json"])
+
+    assert result.exit_code == 0
+    payload = json.loads(result.output)
+    assert payload["zt_pool"]["data"]["tc"] == 2
+    assert payload["dt_pool"]["data"]["tc"] == 1
+    assert payload["zb_pool"]["data"]["tc"] == 3
+
+
+def test_flow_json_outputs_parseable_payload(monkeypatch):
+    from click.testing import CliRunner
+
+    monkeypatch.setattr(cli_module._core, "nearest_trade_date", lambda: "20260529")
+    monkeypatch.setattr(
+        cli_module._core,
+        "get_fund_flow",
+        lambda date: {"主力净流入": "-42899849216.0", "_source": "东财历史日线资金流"},
+    )
+
+    result = CliRunner().invoke(cli, ["flow", "--json"])
+
+    assert result.exit_code == 0
+    payload = json.loads(result.output)
+    assert payload["flow"]["主力净流入"] == "-42899849216.0"
+
+
+def test_global_json_serializes_quote_data(monkeypatch):
+    from click.testing import CliRunner
+
+    quote = cli_module._core.QuoteData(
+        symbol="^HSI",
+        name="恒生指数",
+        market="hk_market",
+        date="20260529",
+        price=18000.0,
+    )
+
+    monkeypatch.setattr(cli_module._core, "nearest_trade_date", lambda: "20260529")
+    monkeypatch.setattr(cli_module._core, "get_index", lambda date: [{"f12": "000001"}])
+    monkeypatch.setattr(cli_module._core, "fetch_hk_indices_tencent", lambda symbols, date: [quote])
+    monkeypatch.setattr(cli_module._core, "fetch_us_indices_sina", lambda symbols, date: [])
+
+    result = CliRunner().invoke(cli, ["global", "--json"])
+
+    assert result.exit_code == 0
+    payload = json.loads(result.output)
+    assert payload["a"][0]["f12"] == "000001"
+    assert payload["hk"][0]["symbol"] == "^HSI"
+
+
 def test_cli_update_runs_pip_upgrade(monkeypatch):
     from click.testing import CliRunner