"""导出 QPro 绩效分析历史数据；仅使用 Python 3 标准库。""" import argparse import csv import json from datetime import datetime from decimal import Decimal from pathlib import Path from urllib.error import HTTPError from urllib.request import Request, urlopen def request(base_url, path, payload=None, method="GET", timeout=60): body = None if payload is None else json.dumps(payload, ensure_ascii=False).encode("utf-8") req = Request(base_url.rstrip("/") + path, data=body, method=method, headers={"Content-Type": "application/json; charset=utf-8"}) try: with urlopen(req, timeout=timeout) as response: result = json.load(response) except HTTPError as error: detail = error.read().decode("utf-8", errors="replace") raise RuntimeError(f"{path}: HTTP {error.code}: {detail}") from error if result.get("code") != 0: raise RuntimeError(f"{path}: {result.get('message', 'invalid response')}") return result["data"] def query_pages(base_url, collection, query, page_size=1000): """查询期间应暂停同步与解析；接口不提供跨页快照。""" records, offset, expected_total = [], 0, None while True: data = request(base_url, f"/{collection}/query", {**query, "limit": page_size, "offset": offset}, method="POST") if expected_total is None: expected_total = data["total"] elif data["total"] != expected_total: raise RuntimeError("分页期间记录总数变化，请等待刷新结束后重新导出") page = data[collection] if not page and offset < expected_total: raise RuntimeError("分页提前返回空页，未取得全部记录") records.extend(page) offset += len(page) if offset >= expected_total: if offset != expected_total: raise RuntimeError("返回记录数量与 total 不一致") return records def write_csv(path, records, empty_fields): fields = list(dict.fromkeys(key for record in records for key in record)) or empty_fields with path.open("w", encoding="utf-8-sig", newline="") as stream: writer = csv.DictWriter(stream, fieldnames=fields) writer.writeheader() writer.writerows(records) # None 输出为空白；原始 JSON 保留 null。 def summarize_fees(accounts, trades): currencies = {(row["trading_day"], row["futures_company_name"], row["client_id"]): row.get("currency") for row in accounts} groups = {} for trade in trades: currency = currencies.get((trade["trading_day"], trade["futures_company_name"], trade["client_id"])) key = (trade["futures_company_name"], trade["client_id"], currency, trade["source_type"], trade.get("exchange"), trade.get("instrument_id")) item = groups.setdefault(key, {"count": 0, "missing": 0, "fee": Decimal("0")}) item["count"] += 1 if trade.get("commission") is None: item["missing"] += 1 else: item["fee"] += Decimal(str(trade["commission"])) fields = ("futures_company_name", "client_id", "currency", "source_type", "exchange", "instrument_id") return [{**dict(zip(fields, key)), "trade_rows": value["count"], "missing_commission_rows": value["missing"], "commission_total": None if value["missing"] else str(value["fee"])} for key, value in groups.items()] def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--base-url", default="http://127.0.0.1:8815") parser.add_argument("--list-accounts", action="store_true") parser.add_argument("--company", help="从 /account-infos 返回值复制期货公司名称") parser.add_argument("--client", help="从 /account-infos 返回值复制客户号，保留前导零") parser.add_argument("--start-date") parser.add_argument("--end-date") parser.add_argument("--refresh", action="store_true", help="先同步并解析新结算数据") parser.add_argument("--output", type=Path, help="新建输出目录；已存在时停止以避免覆盖") args = parser.parse_args() if not args.list_accounts: if not all((args.company, args.client, args.start_date, args.end_date)): parser.error("导出时需要 --company、--client、--start-date 和 --end-date") try: start = datetime.strptime(args.start_date, "%Y-%m-%d") end = datetime.strptime(args.end_date, "%Y-%m-%d") except ValueError: parser.error("日期应为 YYYY-MM-DD") if start > end: parser.error("开始日期不能晚于结束日期") if args.output and args.output.exists(): parser.error("输出目录已存在，请换一个新目录") request(args.base_url, "/health") if args.refresh: for path in ("/settlements/sync", "/settlements/process"): stats = request(args.base_url, path, method="POST", timeout=600) print(path, json.dumps(stats, ensure_ascii=False)) if stats["failed_count"]: raise RuntimeError("本次刷新存在失败事件，请检查后再导出；计数为零也不证明历史数据完整") infos = request(args.base_url, "/account-infos")["accounts"] if args.list_accounts: print(json.dumps(infos, ensure_ascii=False, indent=2)) return selector = {"futures_company_name": args.company, "client_id": args.client} if selector not in infos: raise RuntimeError("账户不在 /account-infos 中，请检查账户选择或先同步并解析数据") query = {"start_date": args.start_date, "end_date": args.end_date, "accounts": [selector]} account_data = request(args.base_url, "/accounts/query", query, method="POST") accounts = account_data["accounts"] if len(accounts) != account_data["total"]: raise RuntimeError("账户返回数量与 total 不一致") trades = query_pages(args.base_url, "trades", query) positions = query_pages(args.base_url, "positions", query) output = args.output or Path("settlement-export-" + datetime.now().strftime("%Y%m%d-%H%M%S-%f")) output.mkdir(parents=True, exist_ok=False) raw = {"query": query, "exported_at": datetime.now().astimezone().isoformat(), "accounts": accounts, "trades": trades, "positions": positions} (output / "records.json").write_text(json.dumps(raw, ensure_ascii=False, indent=2), encoding="utf-8") for name, records in (("accounts", accounts), ("trades", trades), ("positions", positions)): write_csv(output / f"{name}.csv", records, ["trading_day", "futures_company_name", "client_id"]) write_csv(output / "fees_by_contract.csv", summarize_fees(accounts, trades), ["futures_company_name", "client_id", "currency", "source_type", "exchange", "instrument_id", "trade_rows", "missing_commission_rows", "commission_total"]) print(f"已导出到账户数据目录：{output.resolve()}") print(f"账户记录 {len(accounts)} 条；成交 {len(trades)} 条；持仓 {len(positions)} 条") if __name__ == "__main__": main()