diff --git a/cli.py b/cli.py index a6ed9e2..c926e38 100644 --- a/cli.py +++ b/cli.py @@ -20,6 +20,7 @@ from pipeline.alpha.cli import alpha from pipeline.features.cli import feature from pipeline.combo.cli import combo from pipeline.portfolio.cli import portfolio +from plugins.joinquant.cli import joinquant from tools.pqcat import pqcat from tools.alphaview import alphaview @@ -48,6 +49,7 @@ cli.add_command(alpha) cli.add_command(feature) cli.add_command(combo) cli.add_command(portfolio) +cli.add_command(joinquant) cli.add_command(pqcat) cli.add_command(alphaview) diff --git a/docs/joinquant_comparison_plugin.md b/docs/joinquant_comparison_plugin.md new file mode 100644 index 0000000..eb0ffaf --- /dev/null +++ b/docs/joinquant_comparison_plugin.md @@ -0,0 +1,237 @@ +# JoinQuant Comparison Plugin + +## Why a Plugin + +JoinQuant is an external execution and simulation reference. Keeping this code +under `plugins/joinquant/` prevents vendor-specific assumptions from entering +`pipeline/portfolio/`, where the internal reference simulator remains the +canonical implementation. + +## What It Validates + +The comparison is for system correctness: + +- date alignment +- internal to JoinQuant symbol mapping +- target position generation +- once-per-day open execution timing +- lot rounding and filled shares +- position carry +- trading cost +- PnL accounting +- blocked trades from suspension, limit-up, and limit-down conditions + +## What It Does Not Validate + +It does not validate alpha quality, IC, IR, forecast skill, or whether the +strategy is economically useful. Differences can be expected when JoinQuant +uses different fee, slippage, cash, corporate-action, or internal rounding +rules. + +## Historical Backtest Workflow + +```bash +# 1. Build internal portfolio targets. +uv run python cli.py portfolio build ... + +# 2. Export JoinQuant-compatible frozen targets. +uv run python cli.py joinquant export-targets \ + --positions-path portfolio/run1.pq \ + --portfolio-name run1 \ + --mode target_shares \ + --out-dir plugins_output/joinquant/targets + +# 3. Generate and copy the wrapper strategy and target files into JoinQuant. +uv run python cli.py joinquant write-wrapper \ + --portfolio-name run1 \ + --mode target_shares \ + --out-path plugins_output/joinquant/wrapper_strategy_run1.py + +# 4. Run the JoinQuant backtest or simulated trading job. +# 5. Export JoinQuant fills, positions, and daily PnL to CSV. + +# 6. Ingest JoinQuant output. +uv run python cli.py joinquant ingest \ + --portfolio-name run1 \ + --fills-csv path/to/jq_fills.csv \ + --positions-csv path/to/jq_positions.csv \ + --pnl-csv path/to/jq_pnl.csv + +# 7. Reconcile. +uv run python cli.py joinquant reconcile \ + --portfolio-name run1 \ + --targets-dir plugins_output/joinquant/targets/run1 \ + --our-fills-path fills/run1.pq \ + --our-positions-path portfolio/run1.pq \ + --our-pnl-path pnl/run1.pq \ + --jq-fills-path plugins_output/joinquant/ingested/run1/fills.pq \ + --jq-positions-path plugins_output/joinquant/ingested/run1/positions.pq \ + --jq-pnl-path plugins_output/joinquant/ingested/run1/pnl.pq +``` + +## Forward-Testing Workflow + +After the T-1 close and after the data update: + +```bash +uv run python cli.py portfolio build ... +uv run python cli.py joinquant export-targets \ + --positions-path portfolio/run1.pq \ + --portfolio-name run1 \ + --mode target_shares \ + --start-date T \ + --end-date T +``` + +Before the T open, upload or expose the frozen target file to JoinQuant. During +the T open, the JoinQuant wrapper reads that file and submits orders, while the +internal simulator should run against the same frozen target. After T close or +after JoinQuant results are available, ingest the JoinQuant CSV files and run +`joinquant reconcile`. + +Forward target files must be frozen before execution. Do not regenerate a +target file after observing open or close data for the same trading date. The +exporter writes a snapshot JSON with a SHA-256 hash for this reason and refuses +to overwrite existing target/snapshot files unless `--force` is passed. + +## Target-Shares Mode + +`target_shares` is the default and preferred correctness mode. The exported +`target_shares` field comes from the internal `position_shares` column produced +by `portfolio build`, because the internal simulator executes that discretized +integer book. The generated wrapper calls: + +```python +order_target(jq_symbol, target_shares) +``` + +This mode makes filled shares, position carry, and blocked trades easiest to +compare. + +## Target-Value Mode + +`target_value` mode exports `target_value` and `target_weight` from the +portfolio file. The generated wrapper calls: + +```python +order_target_value(jq_symbol, target_value) +``` + +This can be useful for portfolio-level comparisons, but JoinQuant may apply its +own rounding, cash, and lot rules. Differences are often classified as +`JOINQUANT_INTERNAL_ROUNDING`, `LOT_ROUNDING`, or `CASH_CONSTRAINT` depending +on the observed output. + +## Symbol Mapping + +Internal symbols are converted as follows: + +```text +sh600000 -> 600000.XSHG +sh688001 -> 688001.XSHG +sz000001 -> 000001.XSHE +sz300001 -> 300001.XSHE +``` + +Reverse mapping is also supported. Invalid exchanges or unsupported A-share +prefixes raise `ValueError` instead of silently guessing. + +## Wrapper Strategy Usage + +Generate a configured wrapper: + +```bash +uv run python cli.py joinquant write-wrapper \ + --portfolio-name run1 \ + --mode target_shares \ + --out-path plugins_output/joinquant/wrapper_strategy_run1.py +``` + +Copy the generated file and daily CSV target files into JoinQuant. The default +loader uses JoinQuant `read_file`, which works for uploaded files. If your +JoinQuant runtime allows HTTP or another storage backend, replace only +`_read_target_file()` in the generated strategy. + +The wrapper is long-only by default: + +```python +ALLOW_SHORT = False +``` + +Negative targets are clipped to zero and logged. Use `--allow-short` only if +the target JoinQuant account supports the required shorting mechanics. + +## Ingesting JoinQuant Outputs + +The ingest command accepts permissive CSV column names and writes strict plugin +schemas: + +```text +plugins_output/joinquant/ingested/{portfolio_name}/fills.pq +plugins_output/joinquant/ingested/{portfolio_name}/positions.pq +plugins_output/joinquant/ingested/{portfolio_name}/pnl.pq +``` + +Missing cost fields default to zero. Missing blocked status defaults to zero. +Symbols and dates are normalized. + +## Reading Reconciliation Reports + +The reconcile command writes: + +```text +plugins_output/joinquant/reconcile/{portfolio_name}/daily_reconcile.pq +plugins_output/joinquant/reconcile/{portfolio_name}/summary.csv +plugins_output/joinquant/reconcile/{portfolio_name}/summary.md +``` + +`daily_reconcile.pq` is per-symbol and includes target shares, internal filled +shares, JoinQuant filled shares, realized positions, trade prices, costs, PnL, +and a `diff_reason`. `summary.csv` is the daily portfolio-level view for gross +exposure, net exposure, cash, total value, PnL, cumulative PnL, turnover, and +cost. + +Difference reasons include: + +```text +MATCH SYMBOL_MAPPING PRICE_MISMATCH LOT_ROUNDING SUSPENSION LIMIT_UP_BLOCK +LIMIT_DOWN_BLOCK VOLUME_OR_LIQUIDITY COST_MODEL CASH_CONSTRAINT +SHORT_NOT_SUPPORTED CORPORATE_ACTION JOINQUANT_INTERNAL_ROUNDING +MISSING_IN_OUR_SYSTEM MISSING_IN_JOINQUANT UNKNOWN +``` + +Default tolerances are exact share matching, `1e-4` relative trade-price +tolerance, and value tolerance `max(1 yuan, 1e-6 * booksize)`. PnL tolerance is +configurable with `--pnl-tolerance`. + +## Minimal Example + +Create a 5-stock equal-weight or fixed-share test portfolio: + +```text +sh600000, sz000001, sh600519, sz002594, sz300750 +``` + +Build positions for a small date range, export `target_shares`, upload the CSV +files and wrapper to JoinQuant, run the JoinQuant backtest, export fills, +positions, and PnL, then run ingest and reconcile. Start with one or two days +before expanding the sample. + +## Recommended First Sanity Checks + +1. One liquid stock with a fixed target share count. +2. A 10-stock equal-weight long-only portfolio. +3. A forced suspension, limit-up, and limit-down sample. +4. A short target in long-only mode to confirm `SHORT_NOT_SUPPORTED`. +5. A 5-day reversal portfolio after the mechanical checks pass. + +## Known Limitations + +- JoinQuant internal execution details may differ from the reference simulator. +- External file loading depends on the JoinQuant environment. +- Short selling may not be supported. +- Fee, tax, slippage, and minimum-fee models may differ. +- Corporate actions may need special handling and should not be hidden. +- The internal simulator does not currently emit execution price in + `FILL_COLUMNS`; price reconciliation uses explicit price columns if supplied. + diff --git a/plugins/__init__.py b/plugins/__init__.py new file mode 100644 index 0000000..943b042 --- /dev/null +++ b/plugins/__init__.py @@ -0,0 +1,2 @@ +"""Optional plugin packages for the research pipeline.""" + diff --git a/plugins/joinquant/README.md b/plugins/joinquant/README.md new file mode 100644 index 0000000..4da2d47 --- /dev/null +++ b/plugins/joinquant/README.md @@ -0,0 +1,55 @@ +# JoinQuant Comparison Plugin + +This plugin exports frozen targets from the internal A-share research pipeline, +drives a standalone JoinQuant wrapper strategy, ingests JoinQuant output files, +and reconciles them against the internal reference simulator. + +The plugin validates system mechanics, not alpha quality: + +- date alignment +- symbol mapping +- target position generation +- open execution timing +- lot rounding and filled shares +- position carry +- trading cost and PnL accounting +- blocked trades from suspension and price limits + +## Commands + +```bash +uv run python cli.py joinquant export-targets \ + --positions-path portfolio/run1.pq \ + --portfolio-name run1 \ + --mode target_shares \ + --start-date 2026-07-01 \ + --end-date 2026-07-31 \ + --out-dir plugins_output/joinquant/targets + +uv run python cli.py joinquant write-wrapper \ + --portfolio-name run1 \ + --mode target_shares \ + --out-path plugins_output/joinquant/wrapper_strategy_run1.py + +uv run python cli.py joinquant ingest \ + --portfolio-name run1 \ + --fills-csv path/to/jq_fills.csv \ + --positions-csv path/to/jq_positions.csv \ + --pnl-csv path/to/jq_pnl.csv \ + --out-dir plugins_output/joinquant/ingested + +uv run python cli.py joinquant reconcile \ + --portfolio-name run1 \ + --targets-dir plugins_output/joinquant/targets/run1 \ + --our-fills-path fills/run1.pq \ + --our-positions-path portfolio/run1.pq \ + --our-pnl-path pnl/run1.pq \ + --jq-fills-path plugins_output/joinquant/ingested/run1/fills.pq \ + --jq-positions-path plugins_output/joinquant/ingested/run1/positions.pq \ + --jq-pnl-path plugins_output/joinquant/ingested/run1/pnl.pq \ + --out-dir plugins_output/joinquant/reconcile +``` + +`target_shares` is the default and uses the built integer `position_shares` +from `portfolio build`, matching what the internal simulator executes. + diff --git a/plugins/joinquant/__init__.py b/plugins/joinquant/__init__.py new file mode 100644 index 0000000..c4adce3 --- /dev/null +++ b/plugins/joinquant/__init__.py @@ -0,0 +1,10 @@ +"""JoinQuant comparison plugin. + +This package keeps JoinQuant-specific export, ingest, and reconciliation code +outside the core portfolio modules. +""" + +from plugins.joinquant.symbols import from_joinquant_symbol, to_joinquant_symbol + +__all__ = ["from_joinquant_symbol", "to_joinquant_symbol"] + diff --git a/plugins/joinquant/cli.py b/plugins/joinquant/cli.py new file mode 100644 index 0000000..5606c1b --- /dev/null +++ b/plugins/joinquant/cli.py @@ -0,0 +1,142 @@ +"""CLI commands for the JoinQuant comparison plugin.""" + +from __future__ import annotations + +import click + +from plugins.joinquant.export_targets import export_targets +from plugins.joinquant.ingest import ingest_joinquant_outputs +from plugins.joinquant.reconcile import reconcile_joinquant +from plugins.joinquant.wrapper_strategy import write_wrapper_strategy + + +@click.group(name="joinquant") +def joinquant(): + """Compare internal portfolio simulation with JoinQuant output.""" + + +@joinquant.command("export-targets") +@click.option("--positions-path", required=True, help="Portfolio positions parquet from `portfolio build`") +@click.option("--portfolio-name", required=True, help="Portfolio run to export") +@click.option( + "--mode", + "mode", + type=click.Choice(["target_shares", "target_value"]), + default="target_shares", + show_default=True, + help="JoinQuant target order mode", +) +@click.option("--start-date", default=None, help="Inclusive YYYY-MM-DD start date") +@click.option("--end-date", default=None, help="Inclusive YYYY-MM-DD end date") +@click.option("--out-dir", default="plugins_output/joinquant/targets", show_default=True) +@click.option("--force", is_flag=True, help="Overwrite frozen target/snapshot files") +def export_targets_cmd(positions_path, portfolio_name, mode, start_date, end_date, out_dir, force): + """Export frozen daily target files for JoinQuant.""" + snapshots = export_targets( + positions_path=positions_path, + portfolio_name=portfolio_name, + mode=mode, + start_date=start_date, + end_date=end_date, + out_dir=out_dir, + force=force, + ) + click.echo(f"Exported JoinQuant targets: {len(snapshots)} day(s)") + for snapshot in snapshots: + click.echo( + f" {snapshot['date']}: {snapshot['n_symbols']} symbols, " + f"sha256={str(snapshot['file_sha256'])[:12]}" + ) + + +@joinquant.command("ingest") +@click.option("--portfolio-name", required=True, help="Portfolio run name") +@click.option("--fills-csv", required=True, help="JoinQuant fills CSV") +@click.option("--positions-csv", required=True, help="JoinQuant positions CSV") +@click.option("--pnl-csv", required=True, help="JoinQuant daily PnL CSV") +@click.option("--out-dir", default="plugins_output/joinquant/ingested", show_default=True) +def ingest_cmd(portfolio_name, fills_csv, positions_csv, pnl_csv, out_dir): + """Normalize JoinQuant CSV exports to parquet.""" + paths = ingest_joinquant_outputs( + portfolio_name=portfolio_name, + fills_csv=fills_csv, + positions_csv=positions_csv, + pnl_csv=pnl_csv, + out_dir=out_dir, + ) + click.echo(f"Saved JoinQuant fills: {paths['fills']}") + click.echo(f"Saved JoinQuant positions: {paths['positions']}") + click.echo(f"Saved JoinQuant pnl: {paths['pnl']}") + + +@joinquant.command("reconcile") +@click.option("--portfolio-name", required=True, help="Portfolio run name") +@click.option("--targets-dir", required=True, help="Directory containing exported daily target files") +@click.option("--our-fills-path", required=True, help="Internal simulator fills parquet") +@click.option("--our-positions-path", required=True, help="Internal portfolio positions parquet") +@click.option("--our-pnl-path", required=True, help="Internal simulator PnL parquet") +@click.option("--jq-fills-path", required=True, help="Normalized JoinQuant fills parquet") +@click.option("--jq-positions-path", required=True, help="Normalized JoinQuant positions parquet") +@click.option("--jq-pnl-path", required=True, help="Normalized JoinQuant PnL parquet") +@click.option("--out-dir", default="plugins_output/joinquant/reconcile", show_default=True) +@click.option("--share-tolerance", default=0.0, show_default=True, type=float) +@click.option("--price-rel-tolerance", default=1e-4, show_default=True, type=float) +@click.option("--pnl-tolerance", default=1.0, show_default=True, type=float) +@click.option("--booksize", default=None, type=float, help="Booksize for value tolerance inference") +def reconcile_cmd( + portfolio_name, + targets_dir, + our_fills_path, + our_positions_path, + our_pnl_path, + jq_fills_path, + jq_positions_path, + jq_pnl_path, + out_dir, + share_tolerance, + price_rel_tolerance, + pnl_tolerance, + booksize, +): + """Write per-symbol and daily JoinQuant reconciliation reports.""" + paths = reconcile_joinquant( + portfolio_name=portfolio_name, + targets_dir=targets_dir, + our_fills_path=our_fills_path, + our_positions_path=our_positions_path, + our_pnl_path=our_pnl_path, + jq_fills_path=jq_fills_path, + jq_positions_path=jq_positions_path, + jq_pnl_path=jq_pnl_path, + out_dir=out_dir, + share_tolerance=share_tolerance, + price_rel_tolerance=price_rel_tolerance, + pnl_tolerance=pnl_tolerance, + booksize=booksize, + ) + click.echo(f"Saved reconciliation parquet: {paths['daily_reconcile']}") + click.echo(f"Saved reconciliation summary: {paths['summary_md']}") + click.echo(f"Saved reconciliation CSV: {paths['summary_csv']}") + + +@joinquant.command("write-wrapper") +@click.option("--portfolio-name", required=True, help="Portfolio run name") +@click.option( + "--mode", + "mode", + type=click.Choice(["target_shares", "target_value"]), + default="target_shares", + show_default=True, +) +@click.option("--out-path", required=True, help="Path for generated standalone strategy") +@click.option("--allow-short", is_flag=True, help="Do not clip negative targets in the generated wrapper") +def write_wrapper_cmd(portfolio_name, mode, out_path, allow_short): + """Generate a standalone JoinQuant wrapper strategy.""" + path = write_wrapper_strategy( + portfolio_name=portfolio_name, + mode=mode, + out_path=out_path, + allow_short=allow_short, + ) + click.echo(f"Saved JoinQuant wrapper strategy: {path}") + diff --git a/plugins/joinquant/export_targets.py b/plugins/joinquant/export_targets.py new file mode 100644 index 0000000..8e467b0 --- /dev/null +++ b/plugins/joinquant/export_targets.py @@ -0,0 +1,201 @@ +"""Export portfolio positions as frozen JoinQuant target files.""" + +from __future__ import annotations + +import hashlib +import json +import uuid +from datetime import datetime, timezone +from pathlib import Path +from typing import Iterable, Literal + +import pandas as pd + +from pipeline.common.schema import POSITION_COLUMNS +from plugins.joinquant.schema import JOINQUANT_TARGET_COLUMNS +from plugins.joinquant.symbols import to_joinquant_symbol + + +ExportMode = Literal["target_shares", "target_value"] + + +def _date_text(value: object) -> str: + return pd.Timestamp(value).strftime("%Y-%m-%d") + + +def _date_file_stem(date_text: str) -> str: + return pd.Timestamp(date_text).strftime("%Y%m%d") + + +def _snapshot_root_for(targets_root: Path) -> Path: + if targets_root.name == "targets": + return targets_root.parent / "snapshots" + return targets_root / "snapshots" + + +def _sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as fh: + for chunk in iter(lambda: fh.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _check_position_columns(df: pd.DataFrame) -> None: + missing = [col for col in POSITION_COLUMNS if col not in df.columns] + if missing: + raise ValueError(f"Positions input missing required columns: {missing}") + + +def _filter_dates( + df: pd.DataFrame, + start_date: str | None, + end_date: str | None, +) -> pd.DataFrame: + out = df.copy() + out["date"] = pd.to_datetime(out["date"]).dt.normalize() + if start_date: + out = out[out["date"] >= pd.Timestamp(start_date).normalize()] + if end_date: + out = out[out["date"] <= pd.Timestamp(end_date).normalize()] + return out + + +def build_target_frame( + positions: pd.DataFrame, + *, + portfolio_name: str | None = None, + mode: ExportMode = "target_shares", + start_date: str | None = None, + end_date: str | None = None, + snapshot_ids: dict[str, str] | None = None, +) -> pd.DataFrame: + """Build normalized JoinQuant target rows from portfolio positions. + + ``target_shares`` is populated from ``position_shares`` because the core + simulator executes the discretized book, not continuous research shares. + """ + if mode not in {"target_shares", "target_value"}: + raise ValueError("mode must be 'target_shares' or 'target_value'") + _check_position_columns(positions) + + df = _filter_dates(positions, start_date, end_date) + if portfolio_name is not None: + df = df[df["portfolio_name"].astype(str) == portfolio_name] + if df.empty: + return pd.DataFrame(columns=JOINQUANT_TARGET_COLUMNS) + + out = pd.DataFrame({ + "date": df["date"].map(_date_text), + "portfolio_name": df["portfolio_name"].astype(str), + "symbol_id": df["symbol_id"].astype(str), + "jq_symbol": df["symbol_id"].map(to_joinquant_symbol), + "target_shares": pd.to_numeric(df["position_shares"], errors="coerce").fillna(0).astype("int64"), + "target_value": pd.to_numeric(df["target_value"], errors="coerce").fillna(0.0), + "target_weight": pd.to_numeric(df["target_weight"], errors="coerce").fillna(0.0), + "export_mode": mode, + "snapshot_id": "", + }) + + if snapshot_ids: + out["snapshot_id"] = out["date"].map(snapshot_ids).fillna("") + + return out[JOINQUANT_TARGET_COLUMNS].sort_values( + ["date", "portfolio_name", "symbol_id"] + ).reset_index(drop=True) + + +def export_targets( + positions_path: str | Path, + *, + portfolio_name: str, + mode: ExportMode = "target_shares", + out_dir: str | Path = "plugins_output/joinquant/targets", + start_date: str | None = None, + end_date: str | None = None, + force: bool = False, +) -> list[dict[str, object]]: + """Export one daily CSV/parquet target file plus a snapshot JSON per date. + + Args: + positions_path: Parquet file produced by ``portfolio build``. + portfolio_name: Portfolio run to export. + mode: ``target_shares`` or ``target_value``. + out_dir: Target root. Files are written to ``out_dir/portfolio_name``. + If the root is named ``targets``, snapshots are written to the + sibling ``snapshots`` directory. + start_date: Optional inclusive start date. + end_date: Optional inclusive end date. + force: If false, existing target or snapshot files are treated as + frozen and cause ``FileExistsError``. + + Returns: + Snapshot metadata dictionaries, one per exported date. + """ + positions_path = Path(positions_path) + targets_root = Path(out_dir) + snapshot_root = _snapshot_root_for(targets_root) + targets_portfolio_dir = targets_root / portfolio_name + snapshots_portfolio_dir = snapshot_root / portfolio_name + targets_portfolio_dir.mkdir(parents=True, exist_ok=True) + snapshots_portfolio_dir.mkdir(parents=True, exist_ok=True) + + positions = pd.read_parquet(positions_path) + filtered = _filter_dates(positions, start_date, end_date) + filtered = filtered[filtered["portfolio_name"].astype(str) == portfolio_name] + if filtered.empty: + return [] + + date_texts = sorted(filtered["date"].map(_date_text).unique()) + snapshot_ids = { + date_text: f"jq-{portfolio_name}-{date_text}-{uuid.uuid4().hex[:12]}" + for date_text in date_texts + } + targets = build_target_frame( + filtered, + portfolio_name=portfolio_name, + mode=mode, + snapshot_ids=snapshot_ids, + ) + + snapshots: list[dict[str, object]] = [] + for date_text, daily in targets.groupby("date", sort=True): + stem = _date_file_stem(date_text) + csv_path = targets_portfolio_dir / f"{stem}.csv" + parquet_path = targets_portfolio_dir / f"{stem}.parquet" + snapshot_path = snapshots_portfolio_dir / f"{stem}.json" + existing: Iterable[Path] = (csv_path, parquet_path, snapshot_path) + if not force: + conflicts = [str(path) for path in existing if path.exists()] + if conflicts: + raise FileExistsError( + "Frozen JoinQuant target already exists; use --force to overwrite: " + + ", ".join(conflicts) + ) + + daily = daily[JOINQUANT_TARGET_COLUMNS].reset_index(drop=True) + daily.to_csv(csv_path, index=False) + daily.to_parquet(parquet_path, index=False) + file_hash = _sha256_file(csv_path) + + snapshot = { + "snapshot_id": snapshot_ids[date_text], + "portfolio_name": portfolio_name, + "date": date_text, + "export_mode": mode, + "source_positions_path": str(positions_path), + "created_at": datetime.now(timezone.utc).isoformat(), + "n_symbols": int(len(daily)), + "file_sha256": file_hash, + "notes": "Frozen JoinQuant target file.", + "target_csv_path": str(csv_path), + "target_parquet_path": str(parquet_path), + } + snapshot_path.write_text( + json.dumps(snapshot, indent=2, ensure_ascii=False) + "\n", + encoding="utf-8", + ) + snapshots.append(snapshot) + + return snapshots + diff --git a/plugins/joinquant/ingest.py b/plugins/joinquant/ingest.py new file mode 100644 index 0000000..fe0aeeb --- /dev/null +++ b/plugins/joinquant/ingest.py @@ -0,0 +1,225 @@ +"""Normalize JoinQuant CSV exports into plugin parquet schemas.""" + +from __future__ import annotations + +import re +from pathlib import Path + +import numpy as np +import pandas as pd + +from plugins.joinquant.schema import ( + JOINQUANT_FILL_COLUMNS, + JOINQUANT_PNL_COLUMNS, + JOINQUANT_POSITION_COLUMNS, +) +from plugins.joinquant.symbols import normalize_symbol_pair, to_joinquant_symbol + + +def _clean_name(name: object) -> str: + text = str(name).strip().lower() + text = re.sub(r"[\s\-.()/]+", "_", text) + return re.sub(r"[^0-9a-z_]+", "", text).strip("_") + + +def _clean_columns(df: pd.DataFrame) -> pd.DataFrame: + out = df.copy() + out.columns = [_clean_name(col) for col in out.columns] + return out + + +def _pick(df: pd.DataFrame, candidates: list[str]) -> str | None: + for candidate in candidates: + clean = _clean_name(candidate) + if clean in df.columns: + return clean + return None + + +def _series_or_default(df: pd.DataFrame, candidates: list[str], default: object) -> pd.Series: + col = _pick(df, candidates) + if col is None: + return pd.Series([default] * len(df), index=df.index) + return df[col] + + +def _date_series(df: pd.DataFrame) -> pd.Series: + values = _series_or_default( + df, + ["date", "trade_date", "datetime", "time", "created_at"], + pd.NaT, + ) + parsed = pd.to_datetime(values, errors="coerce").dt.normalize() + return parsed.dt.strftime("%Y-%m-%d").fillna("") + + +def _numeric(values: pd.Series, default: float = 0.0) -> pd.Series: + return pd.to_numeric(values, errors="coerce").replace([np.inf, -np.inf], np.nan).fillna(default) + + +def _text(values: pd.Series, default: str = "") -> pd.Series: + return values.fillna(default).astype(str) + + +def _portfolio_series(df: pd.DataFrame, portfolio_name: str) -> pd.Series: + return _text(_series_or_default(df, ["portfolio_name", "portfolio", "strategy"], portfolio_name), portfolio_name) + + +def _symbol_frame(df: pd.DataFrame) -> pd.DataFrame: + internal_col = _pick(df, ["symbol_id", "internal_symbol"]) + jq_col = _pick(df, ["jq_symbol", "security", "stock", "symbol", "code", "order_book_id"]) + + symbol_ids: list[str] = [] + jq_symbols: list[str] = [] + for idx in df.index: + internal = df.at[idx, internal_col] if internal_col else None + jq_value = df.at[idx, jq_col] if jq_col else None + value = internal if internal is not None and str(internal).strip() else jq_value + if value is None or not str(value).strip(): + symbol_ids.append("") + jq_symbols.append("") + continue + symbol_id, jq_symbol = normalize_symbol_pair(value) + if jq_value is not None and str(jq_value).strip(): + try: + _, jq_symbol = normalize_symbol_pair(jq_value) + except ValueError: + jq_symbol = to_joinquant_symbol(symbol_id) + symbol_ids.append(symbol_id) + jq_symbols.append(jq_symbol) + + return pd.DataFrame({"symbol_id": symbol_ids, "jq_symbol": jq_symbols}, index=df.index) + + +def _signed_shares(shares: pd.Series, side: pd.Series) -> pd.Series: + signed = _numeric(shares, 0.0) + side_text = side.fillna("").astype(str).str.lower() + sell = side_text.str.contains("sell|short|close|reduce|-", regex=True) + buy = side_text.str.contains("buy|long|open|add|\\+", regex=True) + signed = signed.abs() + signed = signed.mask(sell, -signed) + signed = signed.mask(~(sell | buy), _numeric(shares, 0.0)) + return signed + + +def normalize_fills_csv(path: str | Path, portfolio_name: str) -> pd.DataFrame: + """Read a JoinQuant fills CSV and return ``JOINQUANT_FILL_COLUMNS``.""" + raw = _clean_columns(pd.read_csv(path)) + symbols = _symbol_frame(raw) + side = _text(_series_or_default(raw, ["side", "action", "direction"], "")) + requested = _signed_shares( + _series_or_default(raw, ["requested_shares", "target_shares", "amount", "order_amount"], 0), + side, + ) + filled = _signed_shares( + _series_or_default(raw, ["filled_shares", "filled", "filled_amount", "deal_amount", "traded_shares"], 0), + side, + ) + price = _numeric(_series_or_default(raw, ["fill_price", "price", "avg_cost", "avg_price"], np.nan), np.nan) + trade_value = _numeric( + _series_or_default(raw, ["trade_value", "value", "filled_value", "turnover"], np.nan), + np.nan, + ) + trade_value = trade_value.fillna((filled * price).abs()).fillna(0.0) + + out = pd.DataFrame({ + "date": _date_series(raw), + "portfolio_name": _portfolio_series(raw, portfolio_name), + "symbol_id": symbols["symbol_id"], + "jq_symbol": symbols["jq_symbol"], + "order_id": _text(_series_or_default(raw, ["order_id", "id"], "")), + "side": side, + "requested_shares": requested.astype(float), + "filled_shares": filled.astype(float), + "fill_price": price.astype(float), + "trade_value": trade_value.astype(float), + "trade_cost": _numeric(_series_or_default(raw, ["trade_cost", "cost", "commission", "fee"], 0.0), 0.0), + "blocked": _numeric(_series_or_default(raw, ["blocked", "is_blocked"], 0), 0).astype("int64"), + "raw_status": _text(_series_or_default(raw, ["raw_status", "status", "order_status"], "")), + }) + return out[JOINQUANT_FILL_COLUMNS] + + +def normalize_positions_csv(path: str | Path, portfolio_name: str) -> pd.DataFrame: + """Read a JoinQuant positions CSV and return ``JOINQUANT_POSITION_COLUMNS``.""" + raw = _clean_columns(pd.read_csv(path)) + symbols = _symbol_frame(raw) + out = pd.DataFrame({ + "date": _date_series(raw), + "portfolio_name": _portfolio_series(raw, portfolio_name), + "symbol_id": symbols["symbol_id"], + "jq_symbol": symbols["jq_symbol"], + "position_shares": _numeric( + _series_or_default(raw, ["position_shares", "shares", "amount", "quantity", "total_amount"], 0), + 0, + ), + "position_value": _numeric( + _series_or_default(raw, ["position_value", "market_value", "value"], 0.0), + 0.0, + ), + "cash": _numeric(_series_or_default(raw, ["cash", "available_cash"], np.nan), np.nan), + "total_value": _numeric( + _series_or_default(raw, ["total_value", "portfolio_value", "total_asset"], np.nan), + np.nan, + ), + }) + return out[JOINQUANT_POSITION_COLUMNS] + + +def normalize_pnl_csv(path: str | Path, portfolio_name: str) -> pd.DataFrame: + """Read a JoinQuant daily PnL CSV and return ``JOINQUANT_PNL_COLUMNS``.""" + raw = _clean_columns(pd.read_csv(path)) + total_value = _numeric( + _series_or_default(raw, ["total_value", "portfolio_value", "total_asset"], np.nan), + np.nan, + ) + pnl = _numeric(_series_or_default(raw, ["pnl", "daily_pnl", "profit", "returns_value"], np.nan), np.nan) + if pnl.isna().all() and total_value.notna().any(): + pnl = total_value.diff().fillna(0.0) + + out = pd.DataFrame({ + "date": _date_series(raw), + "portfolio_name": _portfolio_series(raw, portfolio_name), + "gross_exposure": _numeric( + _series_or_default(raw, ["gross_exposure", "gross", "positions_value", "market_value"], np.nan), + np.nan, + ), + "net_exposure": _numeric( + _series_or_default(raw, ["net_exposure", "net"], np.nan), + np.nan, + ), + "cash": _numeric(_series_or_default(raw, ["cash", "available_cash"], np.nan), np.nan), + "total_value": total_value, + "pnl": pnl.fillna(0.0), + "cost": _numeric(_series_or_default(raw, ["cost", "trade_cost", "commission", "fee"], 0.0), 0.0), + "turnover": _numeric(_series_or_default(raw, ["turnover"], 0.0), 0.0), + }) + return out[JOINQUANT_PNL_COLUMNS] + + +def ingest_joinquant_outputs( + *, + portfolio_name: str, + fills_csv: str | Path, + positions_csv: str | Path, + pnl_csv: str | Path, + out_dir: str | Path = "plugins_output/joinquant/ingested", +) -> dict[str, Path]: + """Normalize JoinQuant CSV exports and write parquet outputs.""" + out_root = Path(out_dir) / portfolio_name + out_root.mkdir(parents=True, exist_ok=True) + + fills = normalize_fills_csv(fills_csv, portfolio_name) + positions = normalize_positions_csv(positions_csv, portfolio_name) + pnl = normalize_pnl_csv(pnl_csv, portfolio_name) + + paths = { + "fills": out_root / "fills.pq", + "positions": out_root / "positions.pq", + "pnl": out_root / "pnl.pq", + } + fills.to_parquet(paths["fills"], index=False) + positions.to_parquet(paths["positions"], index=False) + pnl.to_parquet(paths["pnl"], index=False) + return paths + diff --git a/plugins/joinquant/reconcile.py b/plugins/joinquant/reconcile.py new file mode 100644 index 0000000..3d3cb57 --- /dev/null +++ b/plugins/joinquant/reconcile.py @@ -0,0 +1,607 @@ +"""Reconcile internal simulator output against normalized JoinQuant output.""" + +from __future__ import annotations + +from pathlib import Path + +import numpy as np +import pandas as pd + +from plugins.joinquant.schema import ( + JOINQUANT_FILL_COLUMNS, + JOINQUANT_PNL_COLUMNS, + JOINQUANT_POSITION_COLUMNS, + JOINQUANT_TARGET_COLUMNS, + RECONCILE_COLUMNS, +) +from plugins.joinquant.symbols import to_joinquant_symbol + + +def _date_text(value: object) -> str: + if pd.isna(value): + return "" + return pd.Timestamp(value).strftime("%Y-%m-%d") + + +def _read_parquet(path: str | Path | None) -> pd.DataFrame: + if path is None: + return pd.DataFrame() + return pd.read_parquet(path) + + +def _normalize_common(df: pd.DataFrame, portfolio_name: str) -> pd.DataFrame: + out = df.copy() + if "date" in out.columns: + out["date"] = out["date"].map(_date_text) + else: + out["date"] = "" + if "portfolio_name" not in out.columns: + out["portfolio_name"] = portfolio_name + out["portfolio_name"] = out["portfolio_name"].fillna(portfolio_name).astype(str) + if "symbol_id" in out.columns: + out["symbol_id"] = out["symbol_id"].fillna("").astype(str) + if "jq_symbol" not in out.columns and "symbol_id" in out.columns: + out["jq_symbol"] = out["symbol_id"].map( + lambda s: to_joinquant_symbol(s) if s else "" + ) + elif "jq_symbol" in out.columns: + out["jq_symbol"] = out["jq_symbol"].fillna("").astype(str) + return out + + +def _numeric(df: pd.DataFrame, column: str, default: float = 0.0) -> pd.Series: + if column not in df.columns: + return pd.Series([default] * len(df), index=df.index, dtype=float) + return pd.to_numeric(df[column], errors="coerce").replace([np.inf, -np.inf], np.nan) + + +def _weighted_price(group: pd.DataFrame, price_col: str, shares_col: str) -> float: + prices = pd.to_numeric(group[price_col], errors="coerce") + shares = pd.to_numeric(group[shares_col], errors="coerce").abs() + valid = prices.notna() & shares.notna() & (shares > 0) + if not valid.any(): + return np.nan + return float(np.average(prices[valid], weights=shares[valid])) + + +def _load_targets(targets_dir: str | Path, portfolio_name: str) -> pd.DataFrame: + root = Path(targets_dir) + if not root.exists(): + return pd.DataFrame(columns=JOINQUANT_TARGET_COLUMNS) + + files_by_stem: dict[str, Path] = {} + for path in sorted(root.glob("*.csv")): + files_by_stem[path.stem] = path + for path in sorted(root.glob("*.parquet")): + files_by_stem[path.stem] = path + + frames: list[pd.DataFrame] = [] + for path in files_by_stem.values(): + if path.suffix == ".parquet": + frame = pd.read_parquet(path) + else: + frame = pd.read_csv(path) + frames.append(frame) + + if not frames: + return pd.DataFrame(columns=JOINQUANT_TARGET_COLUMNS) + targets = pd.concat(frames, ignore_index=True) + targets = _normalize_common(targets, portfolio_name) + targets = targets[targets["portfolio_name"].astype(str) == portfolio_name] + if "target_shares" not in targets.columns: + targets["target_shares"] = 0 + return targets.reindex(columns=JOINQUANT_TARGET_COLUMNS) + + +def _aggregate_targets(targets: pd.DataFrame, portfolio_name: str) -> pd.DataFrame: + if targets.empty: + return pd.DataFrame(columns=["date", "portfolio_name", "symbol_id", "jq_symbol", "target_shares"]) + targets = _normalize_common(targets, portfolio_name) + targets["target_shares"] = _numeric(targets, "target_shares", 0.0) + grouped = ( + targets.groupby(["date", "portfolio_name", "symbol_id"], as_index=False) + .agg(jq_symbol=("jq_symbol", "last"), target_shares=("target_shares", "last")) + ) + return grouped + + +def _aggregate_our_fills(fills: pd.DataFrame, portfolio_name: str) -> pd.DataFrame: + if fills.empty: + return pd.DataFrame(columns=[ + "date", "portfolio_name", "symbol_id", "our_filled_shares", + "our_position_shares", "our_cost", "our_trade_price", "our_blocked", + "our_target_shares", + ]) + fills = _normalize_common(fills, portfolio_name) + fills["traded_shares"] = _numeric(fills, "traded_shares", 0.0) + fills["realized_shares"] = _numeric(fills, "realized_shares", np.nan) + fills["trade_cost"] = _numeric(fills, "trade_cost", 0.0).fillna(0.0) + fills["target_shares"] = _numeric(fills, "target_shares", np.nan) + fills["blocked"] = _numeric(fills, "blocked", 0.0).fillna(0.0) + price_col = next( + (col for col in ["trade_price", "fill_price", "execution_price", "price"] if col in fills.columns), + None, + ) + + rows: list[dict[str, object]] = [] + for key, group in fills.groupby(["date", "portfolio_name", "symbol_id"], sort=False): + row = { + "date": key[0], + "portfolio_name": key[1], + "symbol_id": key[2], + "our_filled_shares": float(group["traded_shares"].sum()), + "our_position_shares": float(group["realized_shares"].dropna().iloc[-1]) + if group["realized_shares"].notna().any() else np.nan, + "our_cost": float(group["trade_cost"].sum()), + "our_trade_price": _weighted_price(group, price_col, "traded_shares") + if price_col else np.nan, + "our_blocked": int(group["blocked"].max()), + "our_target_shares": float(group["target_shares"].dropna().iloc[-1]) + if group["target_shares"].notna().any() else np.nan, + } + rows.append(row) + return pd.DataFrame(rows) + + +def _aggregate_our_positions(positions: pd.DataFrame, portfolio_name: str) -> pd.DataFrame: + if positions.empty: + return pd.DataFrame(columns=[ + "date", "portfolio_name", "symbol_id", "jq_symbol", + "our_position_fallback", "our_position_price", + ]) + positions = _normalize_common(positions, portfolio_name) + positions["position_shares"] = _numeric(positions, "position_shares", np.nan) + positions["price"] = _numeric(positions, "price", np.nan) + return ( + positions.groupby(["date", "portfolio_name", "symbol_id"], as_index=False) + .agg( + jq_symbol=("jq_symbol", "last"), + our_position_fallback=("position_shares", "last"), + our_position_price=("price", "last"), + ) + ) + + +def _aggregate_jq_fills(fills: pd.DataFrame, portfolio_name: str) -> pd.DataFrame: + if fills.empty: + return pd.DataFrame(columns=[ + "date", "portfolio_name", "symbol_id", "jq_filled_shares", + "jq_trade_price", "jq_cost", "jq_blocked", "jq_requested_shares", + "raw_status", + ]) + fills = _normalize_common(fills, portfolio_name) + for col in JOINQUANT_FILL_COLUMNS: + if col not in fills.columns: + fills[col] = np.nan + fills["filled_shares"] = _numeric(fills, "filled_shares", 0.0).fillna(0.0) + fills["requested_shares"] = _numeric(fills, "requested_shares", np.nan) + fills["fill_price"] = _numeric(fills, "fill_price", np.nan) + fills["trade_cost"] = _numeric(fills, "trade_cost", 0.0).fillna(0.0) + fills["blocked"] = _numeric(fills, "blocked", 0.0).fillna(0.0) + fills["raw_status"] = fills["raw_status"].fillna("").astype(str) + + rows: list[dict[str, object]] = [] + for key, group in fills.groupby(["date", "portfolio_name", "symbol_id"], sort=False): + row = { + "date": key[0], + "portfolio_name": key[1], + "symbol_id": key[2], + "jq_filled_shares": float(group["filled_shares"].sum()), + "jq_trade_price": _weighted_price(group, "fill_price", "filled_shares"), + "jq_cost": float(group["trade_cost"].sum()), + "jq_blocked": int(group["blocked"].max()), + "jq_requested_shares": float(group["requested_shares"].dropna().iloc[-1]) + if group["requested_shares"].notna().any() else np.nan, + "raw_status": ";".join([s for s in group["raw_status"].astype(str) if s]), + } + rows.append(row) + return pd.DataFrame(rows) + + +def _aggregate_jq_positions(positions: pd.DataFrame, portfolio_name: str) -> pd.DataFrame: + if positions.empty: + return pd.DataFrame(columns=[ + "date", "portfolio_name", "symbol_id", "jq_symbol", "jq_position_shares", + ]) + positions = _normalize_common(positions, portfolio_name) + for col in JOINQUANT_POSITION_COLUMNS: + if col not in positions.columns: + positions[col] = np.nan + positions["position_shares"] = _numeric(positions, "position_shares", np.nan) + return ( + positions.groupby(["date", "portfolio_name", "symbol_id"], as_index=False) + .agg(jq_symbol=("jq_symbol", "last"), jq_position_shares=("position_shares", "last")) + ) + + +def _portfolio_frame(df: pd.DataFrame, portfolio_name: str, prefix: str) -> pd.DataFrame: + if df.empty: + return pd.DataFrame(columns=["date", "portfolio_name"]) + out = _normalize_common(df, portfolio_name) + for col in JOINQUANT_PNL_COLUMNS: + if col not in out.columns: + out[col] = np.nan + keep = ["date", "portfolio_name", "gross_exposure", "net_exposure", "cash", "total_value", "pnl", "cost", "turnover"] + out = out[keep].copy() + for col in keep[2:]: + out[col] = pd.to_numeric(out[col], errors="coerce") + out = out.groupby(["date", "portfolio_name"], as_index=False).last() + return out.rename(columns={col: f"{prefix}_{col}" for col in keep[2:]}) + + +def _infer_booksize(targets: pd.DataFrame, our_pnl: pd.DataFrame, jq_pnl: pd.DataFrame) -> float: + candidates: list[float] = [] + if "target_value" in targets.columns: + gross = ( + pd.to_numeric(targets["target_value"], errors="coerce") + .abs() + .replace([np.inf, -np.inf], np.nan) + .dropna() + .sum() + ) + if gross > 0: + candidates.append(float(gross)) + for df in (our_pnl, jq_pnl): + if "gross_exposure" in df.columns: + val = pd.to_numeric(df["gross_exposure"], errors="coerce").max() + if pd.notna(val) and val > 0: + candidates.append(float(val)) + return max(candidates) if candidates else 1.0 + + +def _status_reason(raw_status: object) -> str | None: + text = str(raw_status or "").lower() + if "suspend" in text or "halt" in text: + return "SUSPENSION" + if "limit_up" in text or "limit up" in text or "up_limit" in text: + return "LIMIT_UP_BLOCK" + if "limit_down" in text or "limit down" in text or "down_limit" in text: + return "LIMIT_DOWN_BLOCK" + if "volume" in text or "liquid" in text: + return "VOLUME_OR_LIQUIDITY" + if "cash" in text or "margin" in text: + return "CASH_CONSTRAINT" + return None + + +def _classify_symbol_row( + row: pd.Series, + *, + share_tol: float, + price_rel_tol: float, + value_tol: float, + pnl_tol: float, +) -> str: + target = row.get("target_shares", np.nan) + filled_diff = abs(row.get("filled_share_diff", 0.0)) + position_diff = abs(row.get("position_share_diff", 0.0)) + our_present = bool(row.get("_our_present", False)) + jq_present = bool(row.get("_jq_present", False)) + + if pd.notna(target) and target < 0 and (filled_diff > share_tol or position_diff > share_tol or not jq_present): + return "SHORT_NOT_SUPPORTED" + if not jq_present and (our_present or pd.notna(target)): + return "MISSING_IN_JOINQUANT" + if not our_present and jq_present: + return "MISSING_IN_OUR_SYSTEM" + + status_reason = _status_reason(row.get("raw_status", "")) + if (filled_diff > share_tol or position_diff > share_tol) and status_reason: + return status_reason + if filled_diff > share_tol or position_diff > share_tol: + return "UNKNOWN" + + our_price = row.get("our_trade_price", np.nan) + jq_price = row.get("jq_trade_price", np.nan) + if pd.notna(our_price) and pd.notna(jq_price): + denom = max(abs(float(our_price)), abs(float(jq_price)), 1.0) + if abs(float(our_price) - float(jq_price)) > price_rel_tol * denom: + return "PRICE_MISMATCH" + + if abs(row.get("cost_diff", 0.0)) > value_tol: + return "COST_MODEL" + if abs(row.get("pnl_diff", 0.0)) > pnl_tol: + return "UNKNOWN" + return "MATCH" + + +def _classify_portfolio_row(row: pd.Series, value_tol: float, pnl_tol: float) -> str: + our_present = bool(row.get("_our_present", False)) + jq_present = bool(row.get("_jq_present", False)) + if not jq_present and our_present: + return "MISSING_IN_JOINQUANT" + if not our_present and jq_present: + return "MISSING_IN_OUR_SYSTEM" + if abs(row.get("cost_diff", 0.0)) > value_tol: + return "COST_MODEL" + if abs(row.get("pnl_diff", 0.0)) > pnl_tol: + return "UNKNOWN" + return "MATCH" + + +def _build_symbol_reconcile( + *, + portfolio_name: str, + targets: pd.DataFrame, + our_fills: pd.DataFrame, + our_positions: pd.DataFrame, + our_pnl: pd.DataFrame, + jq_fills: pd.DataFrame, + jq_positions: pd.DataFrame, + jq_pnl: pd.DataFrame, + share_tol: float, + price_rel_tol: float, + value_tol: float, + pnl_tol: float, +) -> pd.DataFrame: + target_agg = _aggregate_targets(targets, portfolio_name) + our_fill_agg = _aggregate_our_fills(our_fills, portfolio_name) + our_pos_agg = _aggregate_our_positions(our_positions, portfolio_name) + jq_fill_agg = _aggregate_jq_fills(jq_fills, portfolio_name) + jq_pos_agg = _aggregate_jq_positions(jq_positions, portfolio_name) + + key_cols = ["date", "portfolio_name", "symbol_id"] + keys = [] + for frame in [target_agg, our_fill_agg, our_pos_agg, jq_fill_agg, jq_pos_agg]: + if not frame.empty: + keys.append(frame[key_cols]) + if not keys: + return pd.DataFrame(columns=RECONCILE_COLUMNS) + + base = pd.concat(keys, ignore_index=True).drop_duplicates() + result = base.merge(target_agg, on=key_cols, how="left") + result = result.merge(our_fill_agg, on=key_cols, how="left") + result = result.merge(our_pos_agg, on=key_cols, how="left", suffixes=("", "_ourpos")) + result = result.merge(jq_fill_agg, on=key_cols, how="left") + result = result.merge(jq_pos_agg, on=key_cols, how="left", suffixes=("", "_jqpos")) + + jq_symbol_cols = [col for col in result.columns if col.startswith("jq_symbol")] + jq_symbol_values = result[jq_symbol_cols].copy() if jq_symbol_cols else pd.DataFrame(index=result.index) + result["jq_symbol"] = "" + for col in jq_symbol_values.columns: + values = jq_symbol_values[col].fillna("").astype(str) + result["jq_symbol"] = result["jq_symbol"].mask( + result["jq_symbol"].eq("") & values.ne(""), + values, + ) + result["jq_symbol"] = result.apply( + lambda row: row["jq_symbol"] or to_joinquant_symbol(row["symbol_id"]), + axis=1, + ) + + result["_our_present"] = ( + result[["our_filled_shares", "our_position_shares", "our_position_fallback"]] + .notna() + .any(axis=1) + ) + result["_jq_present"] = ( + result[["jq_filled_shares", "jq_position_shares"]].notna().any(axis=1) + ) + + target_shares = pd.to_numeric(result["target_shares"], errors="coerce") + our_target = pd.to_numeric(result["our_target_shares"], errors="coerce") + jq_target = pd.to_numeric(result["jq_requested_shares"], errors="coerce") + result["target_shares"] = target_shares.where(target_shares.notna(), our_target) + result["target_shares"] = result["target_shares"].where( + result["target_shares"].notna(), + jq_target, + ) + + our_position = pd.to_numeric(result["our_position_shares"], errors="coerce") + our_position_fallback = pd.to_numeric(result["our_position_fallback"], errors="coerce") + result["our_position_shares"] = our_position.where( + our_position.notna(), + our_position_fallback, + ) + + for col in [ + "target_shares", "our_filled_shares", "jq_filled_shares", + "our_position_shares", "jq_position_shares", "our_cost", "jq_cost", + ]: + result[col] = pd.to_numeric(result[col], errors="coerce").fillna(0.0) + + result["filled_share_diff"] = result["our_filled_shares"] - result["jq_filled_shares"] + result["position_share_diff"] = result["our_position_shares"] - result["jq_position_shares"] + result["trade_price_diff"] = np.where( + result["our_trade_price"].notna() & result["jq_trade_price"].notna(), + result["our_trade_price"] - result["jq_trade_price"], + np.nan, + ) + result["cost_diff"] = result["our_cost"] - result["jq_cost"] + + our_daily = _portfolio_frame(our_pnl, portfolio_name, "our") + jq_daily = _portfolio_frame(jq_pnl, portfolio_name, "jq") + pnl_daily = our_daily.merge(jq_daily, on=["date", "portfolio_name"], how="outer") + if not pnl_daily.empty: + pnl_daily["our_pnl"] = pd.to_numeric(pnl_daily.get("our_pnl"), errors="coerce").fillna(0.0) + pnl_daily["jq_pnl"] = pd.to_numeric(pnl_daily.get("jq_pnl"), errors="coerce").fillna(0.0) + pnl_daily["pnl_diff"] = pnl_daily["our_pnl"] - pnl_daily["jq_pnl"] + result = result.merge( + pnl_daily[["date", "portfolio_name", "our_pnl", "jq_pnl", "pnl_diff"]], + on=["date", "portfolio_name"], + how="left", + ) + else: + result["our_pnl"] = 0.0 + result["jq_pnl"] = 0.0 + result["pnl_diff"] = 0.0 + for col in ["our_pnl", "jq_pnl", "pnl_diff"]: + result[col] = pd.to_numeric(result[col], errors="coerce").fillna(0.0) + + result["raw_status"] = result.get("raw_status", "").fillna("") + result["diff_reason"] = result.apply( + _classify_symbol_row, + axis=1, + share_tol=share_tol, + price_rel_tol=price_rel_tol, + value_tol=value_tol, + pnl_tol=pnl_tol, + ) + + return result[RECONCILE_COLUMNS].sort_values( + ["date", "portfolio_name", "symbol_id"] + ).reset_index(drop=True) + + +def _build_portfolio_summary( + *, + portfolio_name: str, + our_pnl: pd.DataFrame, + jq_pnl: pd.DataFrame, + value_tol: float, + pnl_tol: float, +) -> pd.DataFrame: + our = _portfolio_frame(our_pnl, portfolio_name, "our") + jq = _portfolio_frame(jq_pnl, portfolio_name, "jq") + if our.empty and jq.empty: + return pd.DataFrame(columns=[ + "date", "portfolio_name", "diff_reason", + "our_pnl", "jq_pnl", "pnl_diff", + ]) + summary = our.merge(jq, on=["date", "portfolio_name"], how="outer") + summary["_our_present"] = summary.filter(regex=r"^our_").notna().any(axis=1) + summary["_jq_present"] = summary.filter(regex=r"^jq_").notna().any(axis=1) + metrics = ["gross_exposure", "net_exposure", "cash", "total_value", "pnl", "cost", "turnover"] + for metric in metrics: + our_col = f"our_{metric}" + jq_col = f"jq_{metric}" + if our_col not in summary.columns: + summary[our_col] = np.nan + if jq_col not in summary.columns: + summary[jq_col] = np.nan + summary[f"{metric}_diff"] = ( + pd.to_numeric(summary[our_col], errors="coerce").fillna(0.0) + - pd.to_numeric(summary[jq_col], errors="coerce").fillna(0.0) + ) + summary["our_cumulative_pnl"] = ( + pd.to_numeric(summary["our_pnl"], errors="coerce").fillna(0.0).cumsum() + ) + summary["jq_cumulative_pnl"] = ( + pd.to_numeric(summary["jq_pnl"], errors="coerce").fillna(0.0).cumsum() + ) + summary["cumulative_pnl_diff"] = summary["our_cumulative_pnl"] - summary["jq_cumulative_pnl"] + summary["diff_reason"] = summary.apply( + _classify_portfolio_row, + axis=1, + value_tol=value_tol, + pnl_tol=pnl_tol, + ) + summary = summary.drop(columns=["_our_present", "_jq_present"]) + return summary.sort_values(["date", "portfolio_name"]).reset_index(drop=True) + + +def _write_summary_md( + path: Path, + *, + portfolio_name: str, + symbol_report: pd.DataFrame, + portfolio_summary: pd.DataFrame, +) -> None: + symbol_counts = ( + symbol_report["diff_reason"].value_counts().sort_index() + if not symbol_report.empty else pd.Series(dtype=int) + ) + portfolio_counts = ( + portfolio_summary["diff_reason"].value_counts().sort_index() + if not portfolio_summary.empty else pd.Series(dtype=int) + ) + lines = [ + "# JoinQuant Reconciliation Summary", + "", + f"Portfolio: `{portfolio_name}`", + "", + "## Per-symbol Difference Counts", + "", + ] + if symbol_counts.empty: + lines.append("No per-symbol rows were produced.") + else: + for reason, count in symbol_counts.items(): + lines.append(f"- {reason}: {int(count)}") + lines.extend(["", "## Daily Portfolio Difference Counts", ""]) + if portfolio_counts.empty: + lines.append("No daily portfolio rows were produced.") + else: + for reason, count in portfolio_counts.items(): + lines.append(f"- {reason}: {int(count)}") + + if not portfolio_summary.empty: + lines.extend(["", "## Daily Portfolio Preview", ""]) + preview_cols = [ + "date", "diff_reason", "our_pnl", "jq_pnl", "pnl_diff", + "our_cost", "jq_cost", "cost_diff", + ] + preview_cols = [col for col in preview_cols if col in portfolio_summary.columns] + lines.append(",".join(preview_cols)) + for row in portfolio_summary[preview_cols].head(20).itertuples(index=False): + lines.append(",".join(str(value) for value in row)) + + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + +def reconcile_joinquant( + *, + portfolio_name: str, + targets_dir: str | Path, + our_fills_path: str | Path, + our_positions_path: str | Path, + our_pnl_path: str | Path, + jq_fills_path: str | Path, + jq_positions_path: str | Path, + jq_pnl_path: str | Path, + out_dir: str | Path = "plugins_output/joinquant/reconcile", + share_tolerance: float = 0.0, + price_rel_tolerance: float = 1e-4, + pnl_tolerance: float = 1.0, + booksize: float | None = None, +) -> dict[str, Path]: + """Reconcile JoinQuant output against internal simulator output.""" + targets = _load_targets(targets_dir, portfolio_name) + our_fills = _read_parquet(our_fills_path) + our_positions = _read_parquet(our_positions_path) + our_pnl = _read_parquet(our_pnl_path) + jq_fills = _read_parquet(jq_fills_path) + jq_positions = _read_parquet(jq_positions_path) + jq_pnl = _read_parquet(jq_pnl_path) + + inferred_booksize = booksize or _infer_booksize(targets, our_pnl, jq_pnl) + value_tol = max(1.0, 1e-6 * float(inferred_booksize)) + + symbol_report = _build_symbol_reconcile( + portfolio_name=portfolio_name, + targets=targets, + our_fills=our_fills, + our_positions=our_positions, + our_pnl=our_pnl, + jq_fills=jq_fills, + jq_positions=jq_positions, + jq_pnl=jq_pnl, + share_tol=share_tolerance, + price_rel_tol=price_rel_tolerance, + value_tol=value_tol, + pnl_tol=pnl_tolerance, + ) + portfolio_summary = _build_portfolio_summary( + portfolio_name=portfolio_name, + our_pnl=our_pnl, + jq_pnl=jq_pnl, + value_tol=value_tol, + pnl_tol=pnl_tolerance, + ) + + root = Path(out_dir) / portfolio_name + root.mkdir(parents=True, exist_ok=True) + paths = { + "daily_reconcile": root / "daily_reconcile.pq", + "summary_csv": root / "summary.csv", + "summary_md": root / "summary.md", + } + symbol_report.to_parquet(paths["daily_reconcile"], index=False) + portfolio_summary.to_csv(paths["summary_csv"], index=False) + _write_summary_md( + paths["summary_md"], + portfolio_name=portfolio_name, + symbol_report=symbol_report, + portfolio_summary=portfolio_summary, + ) + return paths diff --git a/plugins/joinquant/schema.py b/plugins/joinquant/schema.py new file mode 100644 index 0000000..4cc4c84 --- /dev/null +++ b/plugins/joinquant/schema.py @@ -0,0 +1,99 @@ +"""Column contracts for the JoinQuant comparison plugin.""" + +from typing import Final + + +JOINQUANT_TARGET_COLUMNS: Final[list[str]] = [ + "date", + "portfolio_name", + "symbol_id", + "jq_symbol", + "target_shares", + "target_value", + "target_weight", + "export_mode", + "snapshot_id", +] + +JOINQUANT_FILL_COLUMNS: Final[list[str]] = [ + "date", + "portfolio_name", + "symbol_id", + "jq_symbol", + "order_id", + "side", + "requested_shares", + "filled_shares", + "fill_price", + "trade_value", + "trade_cost", + "blocked", + "raw_status", +] + +JOINQUANT_POSITION_COLUMNS: Final[list[str]] = [ + "date", + "portfolio_name", + "symbol_id", + "jq_symbol", + "position_shares", + "position_value", + "cash", + "total_value", +] + +JOINQUANT_PNL_COLUMNS: Final[list[str]] = [ + "date", + "portfolio_name", + "gross_exposure", + "net_exposure", + "cash", + "total_value", + "pnl", + "cost", + "turnover", +] + +RECONCILE_COLUMNS: Final[list[str]] = [ + "date", + "portfolio_name", + "symbol_id", + "jq_symbol", + "target_shares", + "our_filled_shares", + "jq_filled_shares", + "filled_share_diff", + "our_position_shares", + "jq_position_shares", + "position_share_diff", + "our_trade_price", + "jq_trade_price", + "trade_price_diff", + "our_cost", + "jq_cost", + "cost_diff", + "our_pnl", + "jq_pnl", + "pnl_diff", + "diff_reason", +] + +DIFF_REASONS: Final[list[str]] = [ + "MATCH", + "SYMBOL_MAPPING", + "PRICE_MISMATCH", + "LOT_ROUNDING", + "SUSPENSION", + "LIMIT_UP_BLOCK", + "LIMIT_DOWN_BLOCK", + "VOLUME_OR_LIQUIDITY", + "COST_MODEL", + "CASH_CONSTRAINT", + "SHORT_NOT_SUPPORTED", + "CORPORATE_ACTION", + "JOINQUANT_INTERNAL_ROUNDING", + "MISSING_IN_OUR_SYSTEM", + "MISSING_IN_JOINQUANT", + "UNKNOWN", +] + diff --git a/plugins/joinquant/symbols.py b/plugins/joinquant/symbols.py new file mode 100644 index 0000000..4bc2e7b --- /dev/null +++ b/plugins/joinquant/symbols.py @@ -0,0 +1,94 @@ +"""Symbol conversion between internal A-share ids and JoinQuant ids.""" + +from __future__ import annotations + +import re + + +_INTERNAL_RE = re.compile(r"^(?Psh|sz)(?P\d{6})$", re.IGNORECASE) +_JOINQUANT_RE = re.compile( + r"^(?P\d{6})\.(?PXSHG|XSHE)$", re.IGNORECASE +) +_BARE_RE = re.compile(r"^\d{6}$") + + +def _validate_internal(prefix: str, code: str) -> None: + if prefix == "sh" and code.startswith("6"): + return + if prefix == "sz" and code.startswith(("0", "3")): + return + raise ValueError(f"Unsupported A-share symbol: {prefix}{code}") + + +def _validate_joinquant(code: str, exchange: str) -> None: + if exchange == "XSHG" and code.startswith("6"): + return + if exchange == "XSHE" and code.startswith(("0", "3")): + return + raise ValueError(f"Unsupported JoinQuant A-share symbol: {code}.{exchange}") + + +def to_joinquant_symbol(symbol_id: str) -> str: + """Convert an internal symbol like ``sh600000`` to ``600000.XSHG``. + + Args: + symbol_id: Internal A-share id with ``sh`` or ``sz`` prefix. + + Returns: + JoinQuant security id. + + Raises: + ValueError: If the symbol is malformed or outside supported A-share + Shanghai/Shenzhen equity prefixes. + """ + text = str(symbol_id).strip().lower() + match = _INTERNAL_RE.match(text) + if not match: + raise ValueError(f"Invalid internal symbol: {symbol_id!r}") + + prefix = match.group("prefix").lower() + code = match.group("code") + _validate_internal(prefix, code) + exchange = "XSHG" if prefix == "sh" else "XSHE" + return f"{code}.{exchange}" + + +def from_joinquant_symbol(jq_symbol: str) -> str: + """Convert a JoinQuant symbol like ``600000.XSHG`` to ``sh600000``.""" + text = str(jq_symbol).strip().upper() + match = _JOINQUANT_RE.match(text) + if not match: + raise ValueError(f"Invalid JoinQuant symbol: {jq_symbol!r}") + + code = match.group("code") + exchange = match.group("exchange").upper() + _validate_joinquant(code, exchange) + prefix = "sh" if exchange == "XSHG" else "sz" + return f"{prefix}{code}" + + +def normalize_symbol_pair(value: object) -> tuple[str, str]: + """Return ``(symbol_id, jq_symbol)`` from any supported symbol spelling.""" + text = str(value).strip() + if not text or text.lower() == "nan": + raise ValueError("Missing symbol") + + if _INTERNAL_RE.match(text): + symbol_id = text.lower() + return symbol_id, to_joinquant_symbol(symbol_id) + + if _JOINQUANT_RE.match(text): + jq_symbol = text.upper() + return from_joinquant_symbol(jq_symbol), jq_symbol + + if _BARE_RE.match(text): + if text.startswith("6"): + symbol_id = f"sh{text}" + elif text.startswith(("0", "3")): + symbol_id = f"sz{text}" + else: + raise ValueError(f"Unsupported bare A-share code: {text}") + return symbol_id, to_joinquant_symbol(symbol_id) + + raise ValueError(f"Unsupported symbol: {value!r}") + diff --git a/plugins/joinquant/wrapper_strategy.py b/plugins/joinquant/wrapper_strategy.py new file mode 100644 index 0000000..3df5aa4 --- /dev/null +++ b/plugins/joinquant/wrapper_strategy.py @@ -0,0 +1,241 @@ +"""Generate a standalone JoinQuant wrapper strategy. + +The module also defines default JoinQuant strategy hooks by executing the same +template with ``run1`` / ``target_shares`` defaults. That means this file can be +copied directly into JoinQuant for a quick smoke test, while the CLI can still +write a configured standalone file for a real run. +""" + +from __future__ import annotations + +from pathlib import Path +from string import Template +from typing import Literal + + +WrapperMode = Literal["target_shares", "target_value"] + + +_WRAPPER_TEMPLATE = Template( + r'''# Standalone JoinQuant target wrapper generated by chinese-equity-quant. +# Copy this file and the exported daily CSV target files into JoinQuant. +# +# The file loader is isolated in _read_target_file(). The default implementation +# uses JoinQuant's read_file API for uploaded files. If your JoinQuant runtime +# allows HTTP or another storage backend, replace only that function. + +import csv +import io +import json + + +PORTFOLIO_NAME = "${portfolio_name}" +TARGET_MODE = "${mode}" +ALLOW_SHORT = ${allow_short} +TARGET_FILE_PREFIX = "" # Optional uploaded-file prefix, for example "run1/" + + +def initialize(context): + set_benchmark("000300.XSHG") + set_option("use_real_price", True) + g.portfolio_name = PORTFOLIO_NAME + g.target_mode = TARGET_MODE + g.targets_by_date = {} + run_daily(load_targets, time="before_open") + run_daily(rebalance_at_open, time="open") + run_daily(record_after_close, time="after_close") + + +def _today_text(context): + return context.current_dt.strftime("%Y-%m-%d") + + +def _today_file_name(context): + return context.current_dt.strftime("%Y%m%d") + ".csv" + + +def _read_target_file(file_name): + data = read_file(TARGET_FILE_PREFIX + file_name) + if isinstance(data, bytes): + data = data.decode("utf-8") + return data + + +def _load_target_rows(context): + file_name = _today_file_name(context) + text = _read_target_file(file_name) + rows = list(csv.DictReader(io.StringIO(text))) + clean_rows = [] + for row in rows: + if row.get("portfolio_name") and row["portfolio_name"] != PORTFOLIO_NAME: + continue + jq_symbol = row.get("jq_symbol") or row.get("security") or row.get("symbol") + if not jq_symbol: + log.warn("Skipping target row with no jq_symbol: %s" % row) + continue + + if TARGET_MODE == "target_shares": + target = int(float(row.get("target_shares") or 0)) + elif TARGET_MODE == "target_value": + target = float(row.get("target_value") or 0.0) + else: + raise ValueError("Unsupported TARGET_MODE: %s" % TARGET_MODE) + + if not ALLOW_SHORT and target < 0: + log.warn( + "SHORT_NOT_SUPPORTED clipping %s target from %s to 0" % + (jq_symbol, target) + ) + target = 0 + + clean_rows.append({"jq_symbol": jq_symbol, "target": target, "raw": row}) + return clean_rows + + +def load_targets(context): + date_text = _today_text(context) + try: + rows = _load_target_rows(context) + except Exception as exc: + log.error("Failed to load JoinQuant target file for %s: %s" % (date_text, exc)) + rows = [] + g.targets_by_date[date_text] = rows + log.info("JOINQUANT_TARGET_LOAD|%s" % json.dumps({ + "date": date_text, + "portfolio_name": PORTFOLIO_NAME, + "target_mode": TARGET_MODE, + "n_targets": len(rows), + }, sort_keys=True)) + + +def rebalance_at_open(context): + date_text = _today_text(context) + rows = g.targets_by_date.get(date_text, []) + target_symbols = set() + for row in rows: + security = row["jq_symbol"] + target_symbols.add(security) + if TARGET_MODE == "target_shares": + order_target(security, int(row["target"])) + else: + order_target_value(security, float(row["target"])) + log.info("JOINQUANT_ORDER_SUBMIT|%s" % json.dumps({ + "date": date_text, + "portfolio_name": PORTFOLIO_NAME, + "jq_symbol": security, + "target_mode": TARGET_MODE, + "target": row["target"], + }, sort_keys=True)) + + for security in list(context.portfolio.positions.keys()): + if security not in target_symbols: + order_target(security, 0) + log.info("JOINQUANT_ORDER_CLOSE|%s" % json.dumps({ + "date": date_text, + "portfolio_name": PORTFOLIO_NAME, + "jq_symbol": security, + }, sort_keys=True)) + + +def _position_records(context): + records = [] + cash = float(context.portfolio.available_cash) + total_value = float(context.portfolio.total_value) + for security, position in context.portfolio.positions.items(): + records.append({ + "date": _today_text(context), + "portfolio_name": PORTFOLIO_NAME, + "jq_symbol": security, + "position_shares": int(position.total_amount), + "position_value": float(position.value), + "cash": cash, + "total_value": total_value, + }) + return records + + +def _trade_records(context): + records = [] + try: + trades = get_trades() + except Exception: + trades = {} + for trade_id, trade in trades.items(): + amount = int(getattr(trade, "amount", 0)) + price = float(getattr(trade, "price", 0.0)) + security = getattr(trade, "security", "") + side = "buy" if amount >= 0 else "sell" + records.append({ + "date": _today_text(context), + "portfolio_name": PORTFOLIO_NAME, + "jq_symbol": security, + "order_id": str(getattr(trade, "order_id", trade_id)), + "side": side, + "filled_shares": amount, + "fill_price": price, + "trade_value": abs(amount * price), + "trade_cost": float(getattr(trade, "commission", 0.0)), + "raw_status": "filled", + }) + return records + + +def record_after_close(context): + date_text = _today_text(context) + for record in _trade_records(context): + log.info("JOINQUANT_FILL|%s" % json.dumps(record, sort_keys=True)) + for record in _position_records(context): + log.info("JOINQUANT_POSITION|%s" % json.dumps(record, sort_keys=True)) + log.info("JOINQUANT_PNL|%s" % json.dumps({ + "date": date_text, + "portfolio_name": PORTFOLIO_NAME, + "cash": float(context.portfolio.available_cash), + "total_value": float(context.portfolio.total_value), + }, sort_keys=True)) +''' +) + + +# Make this module itself usable as a JoinQuant strategy with defaults. +exec(_WRAPPER_TEMPLATE.substitute( + portfolio_name="run1", + mode="target_shares", + allow_short="False", +)) + + +def render_wrapper_strategy( + *, + portfolio_name: str, + mode: WrapperMode = "target_shares", + allow_short: bool = False, +) -> str: + """Render the standalone JoinQuant wrapper strategy source.""" + if mode not in {"target_shares", "target_value"}: + raise ValueError("mode must be 'target_shares' or 'target_value'") + return _WRAPPER_TEMPLATE.substitute( + portfolio_name=portfolio_name, + mode=mode, + allow_short="True" if allow_short else "False", + ) + + +def write_wrapper_strategy( + *, + portfolio_name: str, + mode: WrapperMode = "target_shares", + out_path: str | Path, + allow_short: bool = False, +) -> Path: + """Write a configured standalone JoinQuant wrapper strategy.""" + path = Path(out_path) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + render_wrapper_strategy( + portfolio_name=portfolio_name, + mode=mode, + allow_short=allow_short, + ), + encoding="utf-8", + ) + return path diff --git a/tests/test_joinquant_plugin.py b/tests/test_joinquant_plugin.py new file mode 100644 index 0000000..fc1552b --- /dev/null +++ b/tests/test_joinquant_plugin.py @@ -0,0 +1,482 @@ +"""Tests for the JoinQuant comparison plugin (network-free).""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +import pandas as pd +import pytest +from click.testing import CliRunner + +from cli import cli +from pipeline.common.schema import FILL_COLUMNS, PNL_COLUMNS, POSITION_COLUMNS +from plugins.joinquant.export_targets import export_targets +from plugins.joinquant.ingest import ( + ingest_joinquant_outputs, + normalize_fills_csv, +) +from plugins.joinquant.reconcile import reconcile_joinquant +from plugins.joinquant.schema import ( + JOINQUANT_FILL_COLUMNS, + JOINQUANT_PNL_COLUMNS, + JOINQUANT_POSITION_COLUMNS, + JOINQUANT_TARGET_COLUMNS, + RECONCILE_COLUMNS, +) +from plugins.joinquant.symbols import from_joinquant_symbol, to_joinquant_symbol +from plugins.joinquant.wrapper_strategy import write_wrapper_strategy + + +def _positions( + *, + symbol: str = "sh600000", + date: str = "2026-07-01", + shares: int = 1000, + price: float = 10.0, + portfolio_name: str = "run1", +) -> pd.DataFrame: + target_value = float(shares * price) + weight = target_value / 1_000_000.0 + return pd.DataFrame([{ + "symbol_id": symbol, + "date": pd.Timestamp(date), + "portfolio_name": portfolio_name, + "target_weight": weight, + "target_value": target_value, + "target_shares": float(shares) + 0.25, + "position_shares": shares, + "position_value": target_value, + "price": price, + }], columns=POSITION_COLUMNS) + + +def _our_fills( + *, + symbol: str = "sh600000", + date: str = "2026-07-01", + shares: int = 1000, + price: float = 10.0, + cost: float = 5.0, + portfolio_name: str = "run1", +) -> pd.DataFrame: + fills = pd.DataFrame([{ + "symbol_id": symbol, + "date": pd.Timestamp(date), + "portfolio_name": portfolio_name, + "prev_shares": 0, + "target_shares": shares, + "traded_shares": shares, + "realized_shares": shares, + "blocked": 0, + "trade_cost": cost, + "trade_price": price, + }]) + return fills + + +def _our_pnl( + *, + date: str = "2026-07-01", + pnl: float = 100.0, + cost: float = 5.0, + portfolio_name: str = "run1", +) -> pd.DataFrame: + return pd.DataFrame([{ + "date": pd.Timestamp(date), + "portfolio_name": portfolio_name, + "gross_exposure": 10_000.0, + "net_exposure": 10_000.0, + "pnl": pnl, + "cost": cost, + "turnover": 1.0, + "n_positions": 1, + }], columns=PNL_COLUMNS) + + +def _jq_fills( + *, + symbol: str = "sh600000", + date: str = "2026-07-01", + shares: int = 1000, + price: float = 10.0, + cost: float = 5.0, + portfolio_name: str = "run1", + raw_status: str = "filled", +) -> pd.DataFrame: + return pd.DataFrame([{ + "date": date, + "portfolio_name": portfolio_name, + "symbol_id": symbol, + "jq_symbol": to_joinquant_symbol(symbol), + "order_id": "ord-1", + "side": "buy" if shares >= 0 else "sell", + "requested_shares": shares, + "filled_shares": shares, + "fill_price": price, + "trade_value": abs(shares * price), + "trade_cost": cost, + "blocked": 0, + "raw_status": raw_status, + }], columns=JOINQUANT_FILL_COLUMNS) + + +def _jq_positions( + *, + symbol: str = "sh600000", + date: str = "2026-07-01", + shares: int = 1000, + price: float = 10.0, + portfolio_name: str = "run1", +) -> pd.DataFrame: + return pd.DataFrame([{ + "date": date, + "portfolio_name": portfolio_name, + "symbol_id": symbol, + "jq_symbol": to_joinquant_symbol(symbol), + "position_shares": shares, + "position_value": shares * price, + "cash": 990_000.0, + "total_value": 1_000_000.0, + }], columns=JOINQUANT_POSITION_COLUMNS) + + +def _jq_pnl( + *, + date: str = "2026-07-01", + pnl: float = 100.0, + cost: float = 5.0, + portfolio_name: str = "run1", +) -> pd.DataFrame: + return pd.DataFrame([{ + "date": date, + "portfolio_name": portfolio_name, + "gross_exposure": 10_000.0, + "net_exposure": 10_000.0, + "cash": 990_000.0, + "total_value": 1_000_000.0, + "pnl": pnl, + "cost": cost, + "turnover": 1.0, + }], columns=JOINQUANT_PNL_COLUMNS) + + +def _write_parquets(tmp_path: Path, frames: dict[str, pd.DataFrame]) -> dict[str, Path]: + paths = {} + for name, frame in frames.items(): + path = tmp_path / f"{name}.pq" + frame.to_parquet(path, index=False) + paths[name] = path + return paths + + +def _export_targets_for(tmp_path: Path, positions: pd.DataFrame) -> tuple[Path, Path]: + positions_path = tmp_path / "positions.pq" + positions.to_parquet(positions_path, index=False) + targets_root = tmp_path / "targets" + export_targets( + positions_path, + portfolio_name="run1", + out_dir=targets_root, + mode="target_shares", + ) + return positions_path, targets_root / "run1" + + +@pytest.mark.parametrize( + ("internal", "joinquant"), + [ + ("sh600000", "600000.XSHG"), + ("sh688001", "688001.XSHG"), + ("sz000001", "000001.XSHE"), + ("sz001001", "001001.XSHE"), + ("sz002594", "002594.XSHE"), + ("sz300001", "300001.XSHE"), + ], +) +def test_symbol_mapping_both_directions(internal, joinquant): + assert to_joinquant_symbol(internal) == joinquant + assert from_joinquant_symbol(joinquant) == internal + + +@pytest.mark.parametrize("bad", ["600000", "bj830000", "sh000001", "sz600000", "abc"]) +def test_symbol_mapping_rejects_invalid_symbols(bad): + with pytest.raises(ValueError): + to_joinquant_symbol(bad) + + +@pytest.mark.parametrize("bad", ["600000", "600000.XSHE", "000001.XSHG", "abc.XSHG"]) +def test_reverse_symbol_mapping_rejects_invalid_symbols(bad): + with pytest.raises(ValueError): + from_joinquant_symbol(bad) + + +def test_export_targets_schema_snapshot_hash_and_no_overwrite(tmp_path): + positions_path = tmp_path / "positions.pq" + _positions().to_parquet(positions_path, index=False) + + snapshots = export_targets( + positions_path, + portfolio_name="run1", + out_dir=tmp_path / "targets", + mode="target_shares", + ) + + csv_path = tmp_path / "targets" / "run1" / "20260701.csv" + parquet_path = tmp_path / "targets" / "run1" / "20260701.parquet" + snapshot_path = tmp_path / "snapshots" / "run1" / "20260701.json" + assert csv_path.exists() + assert parquet_path.exists() + assert snapshot_path.exists() + + target = pd.read_csv(csv_path) + assert list(target.columns) == JOINQUANT_TARGET_COLUMNS + assert int(target.loc[0, "target_shares"]) == 1000 + assert float(target.loc[0, "target_value"]) == 10_000.0 + assert target.loc[0, "export_mode"] == "target_shares" + + snapshot = json.loads(snapshot_path.read_text()) + actual_hash = hashlib.sha256(csv_path.read_bytes()).hexdigest() + assert snapshots[0]["file_sha256"] == actual_hash + assert snapshot["file_sha256"] == actual_hash + assert snapshot["n_symbols"] == 1 + + with pytest.raises(FileExistsError): + export_targets( + positions_path, + portfolio_name="run1", + out_dir=tmp_path / "targets", + mode="target_shares", + ) + + +def test_export_targets_target_value_mode_from_position_columns(tmp_path): + positions_path = tmp_path / "positions.pq" + _positions(shares=250, price=20.0).to_parquet(positions_path, index=False) + + export_targets( + positions_path, + portfolio_name="run1", + out_dir=tmp_path / "targets_value", + mode="target_value", + ) + + target = pd.read_parquet(tmp_path / "targets_value" / "run1" / "20260701.parquet") + assert list(target.columns) == JOINQUANT_TARGET_COLUMNS + assert target.loc[0, "export_mode"] == "target_value" + assert target.loc[0, "target_value"] == 5_000.0 + assert target.loc[0, "target_shares"] == 250 + + +def test_ingest_permissive_csv_column_mapping_and_output_schemas(tmp_path): + fills_csv = tmp_path / "jq_fills.csv" + positions_csv = tmp_path / "jq_positions.csv" + pnl_csv = tmp_path / "jq_pnl.csv" + pd.DataFrame([{ + "Trade Date": "2026-07-01 09:31:00", + "Security": "600000.XSHG", + "Direction": "buy", + "Order Amount": 1000, + "Filled Amount": 1000, + "Price": 10.0, + "Status": "filled", + }]).to_csv(fills_csv, index=False) + pd.DataFrame([{ + "Date": "2026-07-01", + "Security": "600000.XSHG", + "Shares": 1000, + "Market Value": 10_000.0, + "Cash": 990_000.0, + "Portfolio Value": 1_000_000.0, + }]).to_csv(positions_csv, index=False) + pd.DataFrame([{ + "Date": "2026-07-01", + "Portfolio Value": 1_000_000.0, + "Daily PnL": 100.0, + "Turnover": 1.0, + }]).to_csv(pnl_csv, index=False) + + fills = normalize_fills_csv(fills_csv, "run1") + assert list(fills.columns) == JOINQUANT_FILL_COLUMNS + assert fills.loc[0, "symbol_id"] == "sh600000" + assert fills.loc[0, "jq_symbol"] == "600000.XSHG" + assert fills.loc[0, "trade_cost"] == 0.0 + assert fills.loc[0, "blocked"] == 0 + + paths = ingest_joinquant_outputs( + portfolio_name="run1", + fills_csv=fills_csv, + positions_csv=positions_csv, + pnl_csv=pnl_csv, + out_dir=tmp_path / "ingested", + ) + assert list(pd.read_parquet(paths["fills"]).columns) == JOINQUANT_FILL_COLUMNS + assert list(pd.read_parquet(paths["positions"]).columns) == JOINQUANT_POSITION_COLUMNS + assert list(pd.read_parquet(paths["pnl"]).columns) == JOINQUANT_PNL_COLUMNS + + +def _run_reconcile_case( + tmp_path: Path, + *, + positions: pd.DataFrame | None = None, + our_fills: pd.DataFrame | None = None, + jq_fills: pd.DataFrame | None = None, + jq_positions: pd.DataFrame | None = None, + our_pnl: pd.DataFrame | None = None, + jq_pnl: pd.DataFrame | None = None, +) -> pd.DataFrame: + positions = _positions() if positions is None else positions + _, targets_dir = _export_targets_for(tmp_path, positions) + paths = _write_parquets(tmp_path, { + "our_fills": _our_fills() if our_fills is None else our_fills, + "our_positions": positions, + "our_pnl": _our_pnl() if our_pnl is None else our_pnl, + "jq_fills": _jq_fills() if jq_fills is None else jq_fills, + "jq_positions": _jq_positions() if jq_positions is None else jq_positions, + "jq_pnl": _jq_pnl() if jq_pnl is None else jq_pnl, + }) + out_paths = reconcile_joinquant( + portfolio_name="run1", + targets_dir=targets_dir, + our_fills_path=paths["our_fills"], + our_positions_path=paths["our_positions"], + our_pnl_path=paths["our_pnl"], + jq_fills_path=paths["jq_fills"], + jq_positions_path=paths["jq_positions"], + jq_pnl_path=paths["jq_pnl"], + out_dir=tmp_path / "reconcile", + ) + report = pd.read_parquet(out_paths["daily_reconcile"]) + assert list(report.columns) == RECONCILE_COLUMNS + assert out_paths["summary_md"].exists() + assert out_paths["summary_csv"].exists() + return report + + +def test_reconcile_exact_match(tmp_path): + report = _run_reconcile_case(tmp_path) + assert report.loc[0, "diff_reason"] == "MATCH" + assert report.loc[0, "filled_share_diff"] == 0 + assert report.loc[0, "position_share_diff"] == 0 + + +def test_reconcile_price_mismatch(tmp_path): + report = _run_reconcile_case(tmp_path, jq_fills=_jq_fills(price=10.5)) + assert report.loc[0, "diff_reason"] == "PRICE_MISMATCH" + + +def test_reconcile_cost_mismatch(tmp_path): + report = _run_reconcile_case( + tmp_path, + jq_fills=_jq_fills(cost=8.0), + jq_pnl=_jq_pnl(cost=8.0), + ) + assert report.loc[0, "diff_reason"] == "COST_MODEL" + + +def test_reconcile_missing_symbol_in_joinquant(tmp_path): + empty_jq_fills = pd.DataFrame(columns=JOINQUANT_FILL_COLUMNS) + empty_jq_positions = pd.DataFrame(columns=JOINQUANT_POSITION_COLUMNS) + report = _run_reconcile_case( + tmp_path, + jq_fills=empty_jq_fills, + jq_positions=empty_jq_positions, + ) + assert report.loc[0, "diff_reason"] == "MISSING_IN_JOINQUANT" + + +def test_reconcile_short_target_with_long_only_joinquant_output(tmp_path): + positions = _positions(shares=-100, price=10.0) + our_fills = _our_fills(shares=-100, price=10.0) + jq_fills = _jq_fills(shares=0, price=10.0, cost=0.0, raw_status="short clipped") + jq_positions = _jq_positions(shares=0, price=10.0) + + report = _run_reconcile_case( + tmp_path, + positions=positions, + our_fills=our_fills, + jq_fills=jq_fills, + jq_positions=jq_positions, + ) + assert report.loc[0, "diff_reason"] == "SHORT_NOT_SUPPORTED" + + +def test_joinquant_cli_smoke_export_ingest_reconcile_and_wrapper(tmp_path): + runner = CliRunner() + positions_path = tmp_path / "positions.pq" + _positions().to_parquet(positions_path, index=False) + + result = runner.invoke(cli, [ + "joinquant", "export-targets", + "--positions-path", str(positions_path), + "--portfolio-name", "run1", + "--mode", "target_shares", + "--out-dir", str(tmp_path / "targets"), + ]) + assert result.exit_code == 0, result.output + assert "Exported JoinQuant targets" in result.output + + fills_csv = tmp_path / "jq_fills.csv" + positions_csv = tmp_path / "jq_positions.csv" + pnl_csv = tmp_path / "jq_pnl.csv" + _jq_fills().to_csv(fills_csv, index=False) + _jq_positions().to_csv(positions_csv, index=False) + _jq_pnl().to_csv(pnl_csv, index=False) + + result = runner.invoke(cli, [ + "joinquant", "ingest", + "--portfolio-name", "run1", + "--fills-csv", str(fills_csv), + "--positions-csv", str(positions_csv), + "--pnl-csv", str(pnl_csv), + "--out-dir", str(tmp_path / "ingested"), + ]) + assert result.exit_code == 0, result.output + assert "Saved JoinQuant fills" in result.output + + paths = _write_parquets(tmp_path, { + "our_fills": _our_fills(), + "our_pnl": _our_pnl(), + }) + result = runner.invoke(cli, [ + "joinquant", "reconcile", + "--portfolio-name", "run1", + "--targets-dir", str(tmp_path / "targets" / "run1"), + "--our-fills-path", str(paths["our_fills"]), + "--our-positions-path", str(positions_path), + "--our-pnl-path", str(paths["our_pnl"]), + "--jq-fills-path", str(tmp_path / "ingested" / "run1" / "fills.pq"), + "--jq-positions-path", str(tmp_path / "ingested" / "run1" / "positions.pq"), + "--jq-pnl-path", str(tmp_path / "ingested" / "run1" / "pnl.pq"), + "--out-dir", str(tmp_path / "reconcile"), + ]) + assert result.exit_code == 0, result.output + assert "Saved reconciliation parquet" in result.output + + wrapper_path = tmp_path / "wrapper_strategy_run1.py" + result = runner.invoke(cli, [ + "joinquant", "write-wrapper", + "--portfolio-name", "run1", + "--mode", "target_shares", + "--out-path", str(wrapper_path), + ]) + assert result.exit_code == 0, result.output + assert "Saved JoinQuant wrapper strategy" in result.output + text = wrapper_path.read_text() + assert 'PORTFOLIO_NAME = "run1"' in text + assert 'TARGET_MODE = "target_shares"' in text + assert "ALLOW_SHORT = False" in text + + +def test_wrapper_strategy_generation_smoke(tmp_path): + path = write_wrapper_strategy( + portfolio_name="run2", + mode="target_value", + out_path=tmp_path / "wrapper.py", + ) + text = path.read_text() + assert 'PORTFOLIO_NAME = "run2"' in text + assert 'TARGET_MODE = "target_value"' in text + assert "order_target_value" in text +