feat: contracts portfolio CLI

.planning/ROADMAP.md

@@ -46,6 +46,7 @@
 - [x] `financing_cost_estimate()` 已支持保证金资金成本估算。
 - [x] `position_pnl_estimate()` 已输出扣减资金成本后的净 P&L。
 - [x] `summarize_batch_portfolio()` 已提供批量持仓高层汇总。
+- [x] `contracts portfolio` CLI 已接入批量持仓高层汇总。
 - [ ] 在完整组合资金引擎中接入融资/资金费率与更多组合约束。
 - [x] T+1 close 执行近似已加入 `backtest --t1-close`（仍非 open/settle，真实缺口待合约数据接入）。
 - [x] `summarize_roll_costs()` 已提供合约级可量化换月 gap 成本统计。
@@ -76,4 +77,4 @@
 
 当前测试：`PYTHONPATH=src python3 -m unittest discover -s tests -v`
 
-当前通过：106 tests OK。
+当前通过：107 tests OK。

CHANGELOG.md

@@ -52,7 +52,8 @@
 - `check_portfolio_constraints()`：组合级保证金占用与回撤风控门控；
 - `financing_cost_estimate()`：保证金资金成本估算；
 - `position_pnl_estimate()` 增加资金成本与净 P&L 估算；
-- `summarize_batch_portfolio()`：批量 episode → 持仓模拟 → 资金/保证金高层汇总。
+- `summarize_batch_portfolio()`：批量 episode → 持仓模拟 → 资金/保证金高层汇总；
+- `contracts portfolio` CLI：从合约 CSV 直接运行批量持仓高层汇总。
 
 ## [0.1.0-rc1] - 2026-09-02
 

README.md

@@ -159,6 +159,7 @@ date,instrument,symbol,contract_month,close,volume,open_interest,open,settle
 ```bash
 goratio contracts inspect --csv /path/to/contracts.csv --json
 goratio contracts backtest --csv /path/to/contracts.csv --roll-adjusted --json
+goratio contracts portfolio --csv /path/to/contracts.csv --json
 ```
 
 同一份合约 CSV 也可桥接为现有研究管线可消费的 `RawMarketData`，从而用真实主力合约链而不是连续指数进入 `analyze`/`backtest` 流程。另有 `build_roll_adjusted_series()` 可把主力链调整为“换月无虚假跳空”的连续收益序列，供跨换月收益计算使用；其调整后绝对价格不能用于真实金油比水平。桥接函数也支持 `roll_adjusted=True`，直接生成带换月无跳空调整的 RawMarketData。`roll_aware_contract_return()` 则显式模拟换月结算与重开仓，可用于验证真实合约链持有期收益。`contract_episode_return_summary()` 可将一组 episode 的连续序列收益与真实换月收益对比。若合约 CSV 提供 `open`/`settle`，`t1_open_settle_gap()` 可计算信号日收盘到下一交易日 open/settle 的执行缺口；episode 对比摘要也会输出 T+1 open gap 统计，并给出扣减 open gap 后的多头真实换月净收益。

src/goratio/cli.py

@@ -23,6 +23,7 @@
     contract_csv_to_raw_market_data,
     read_contract_csv,
 )
+from .margin import summarize_batch_portfolio
 from .dataset import DataQualityError, prepare_market_data
 from .episode_study import run_episode_evidence_bundle
 from .formal_v2 import generate_v2_formal_report
@@ -123,6 +124,14 @@ def build_parser() -> argparse.ArgumentParser:
     contracts_backtest.add_argument("--roll-cost-bps", type=float, default=0.0)
     contracts_backtest.add_argument("--json", action="store_true", dest="as_json")
 
+    contracts_portfolio = contracts_subcommands.add_parser("portfolio", help="用合约 CSV 运行批量持仓高层汇总")
+    contracts_portfolio.add_argument("--csv", type=str, required=True)
+    contracts_portfolio.add_argument("--horizon", type=int, choices=(63,126,252), default=126)
+    contracts_portfolio.add_argument("--direction", type=int, default=1)
+    contracts_portfolio.add_argument("--lots", type=int, default=1)
+    contracts_portfolio.add_argument("--initial-capital", type=float, default=100000.0)
+    contracts_portfolio.add_argument("--json", action="store_true", dest="as_json")
+
     plugin = commands.add_parser("plugin", help="查看静态插件白名单")
     plugin_subcommands = plugin.add_subparsers(dest="plugin_command", required=True)
     plugin_list = plugin_subcommands.add_parser("list", help="列出白名单插件")
@@ -406,6 +415,45 @@ def main(
                     f"{_format_number(report['metrics']['mean_net_return'])}\n"
                 )
             return 0
+        if args.command == "contracts" and args.contracts_command == "portfolio":
+            records = read_contract_csv(args.csv)
+            raw = contract_csv_to_raw_market_data(args.csv)
+            prepared = prepare_market_data(
+                raw,
+                period="10y",
+                completed_before=completed_before,
+                provenance="user_contract_csv",
+                cache_stale=False,
+            )
+            episodes = build_forward_episodes(
+                prepared.history.points,
+                prepared.selected.points,
+                horizon=args.horizon,
+            )
+            report = summarize_batch_portfolio(
+                records,
+                episodes,
+                direction=args.direction,
+                lots=args.lots,
+                initial_capital=args.initial_capital,
+            )
+            if args.as_json:
+                stdout.write(
+                    json.dumps(
+                        report,
+                        ensure_ascii=False,
+                        indent=2,
+                        sort_keys=True,
+                        allow_nan=False,
+                    )
+                    + "\n"
+                )
+            else:
+                stdout.write(
+                    f"合约 CSV 批量组合：episode {report['simulation']['episode_count']}；"
+                    f"最终权益 {_format_number(report['equity']['final_equity'], 2)}\n"
+                )
+            return 0
         if args.command == "contracts":
             records = read_contract_csv(args.csv)
             report = build_contract_series(records)

tests/test_cli.py

@@ -371,5 +371,31 @@ def test_formal_command_returns_v2_report(self) -> None:
         self.assertEqual(payload["protocol"], "goratio-2a-v1")
         self.assertEqual(payload["overall_status"], "insufficient_data")
 
+    def test_contracts_portfolio_runs_from_contract_csv(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            csv_path = Path(directory) / "contracts.csv"
+            csv_path.write_text(
+                "date,instrument,symbol,contract_month,close,volume,open_interest\n"
+                "2024-01-02,gold,GC,2024-02,2000,100,50\n"
+                "2024-01-03,gold,GC,2024-02,2010,80,40\n"
+                "2024-01-02,oil,CL,2024-03,75,1000,500\n"
+                "2024-01-03,oil,CL,2024-03,80,900,400\n",
+                encoding="utf-8",
+            )
+            output = io.StringIO()
+            exit_code = main(
+                [
+                    "contracts", "portfolio", "--csv", str(csv_path),
+                    "--horizon", "63", "--json",
+                ],
+                today=lambda: date(2024, 1, 4),
+                stdout=output,
+                stderr=io.StringIO(),
+            )
+
+        payload = json.loads(output.getvalue())
+        self.assertEqual(exit_code, 0)
+        self.assertEqual(payload["simulation"]["episode_count"], 0)
+
 if __name__ == "__main__":
     unittest.main()