Yuxuan Yan
711 lines
27 KiB
Python
711 lines
27 KiB
Python
"""Browser automation for JoinQuant cloud backtest and simulated-trading runs.
|
|
|
|
JoinQuant's public ``jqdatasdk`` is a data API; cloud strategy upload/run/export
|
|
is exposed through the web application. This module keeps that automation
|
|
optional and configurable so UI selector drift does not affect the core test
|
|
suite.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from plugins.joinquant.ingest import ingest_joinquant_outputs
|
|
from plugins.joinquant.reconcile import reconcile_joinquant
|
|
|
|
|
|
DEFAULT_LOGIN_URL = "https://www.joinquant.com/user/login/index"
|
|
|
|
|
|
def _require_playwright():
|
|
try:
|
|
from playwright.sync_api import TimeoutError as PlaywrightTimeoutError
|
|
from playwright.sync_api import sync_playwright
|
|
except ImportError as exc:
|
|
raise RuntimeError(
|
|
"Playwright is required for JoinQuant browser automation. Install it "
|
|
"in this uv environment, then install Chromium:\n"
|
|
" uv sync --extra joinquant-browser\n"
|
|
" uv run playwright install chromium"
|
|
) from exc
|
|
return sync_playwright, PlaywrightTimeoutError
|
|
|
|
|
|
def _backtest_actions() -> list[dict[str, Any]]:
|
|
return [
|
|
{"type": "goto", "url": "{strategy_url}"},
|
|
{
|
|
"type": "click",
|
|
"selector": "text=不再提示",
|
|
"timeout_ms": 5000,
|
|
"force": True,
|
|
"optional": True,
|
|
"description": "Dismiss JoinQuant first-run editor guide.",
|
|
},
|
|
{
|
|
"type": "click",
|
|
"selector": "text=确定",
|
|
"timeout_ms": 5000,
|
|
"force": True,
|
|
"optional": True,
|
|
"description": "Close JoinQuant first-run editor guide.",
|
|
},
|
|
{
|
|
"type": "click",
|
|
"selector": "text=跳过",
|
|
"timeout_ms": 5000,
|
|
"force": True,
|
|
"optional": True,
|
|
"description": "Skip JoinQuant first-run editor guide.",
|
|
},
|
|
{
|
|
"type": "evaluate",
|
|
"script": (
|
|
"() => {"
|
|
"document.querySelectorAll("
|
|
"'.introjs-overlay,.introjs-helperLayer,.introjs-tooltipReferenceLayer,"
|
|
".introjs-tooltip,.introjs-disableInteraction'"
|
|
").forEach((node) => node.remove());"
|
|
"document.body.classList.remove('introjs-noscroll');"
|
|
"}"
|
|
),
|
|
"optional": True,
|
|
"description": "Remove any remaining JoinQuant intro overlay.",
|
|
},
|
|
{
|
|
"type": "set_ace_editor_file",
|
|
"selector": ".ace_editor",
|
|
"path": "{wrapper_path}",
|
|
"description": "Set generated wrapper strategy in the Ace code editor.",
|
|
},
|
|
{
|
|
"type": "set_input_files",
|
|
"selector": "input[type=file]",
|
|
"paths": "{target_csvs}",
|
|
"description": "Upload all aligned daily target CSV files.",
|
|
"timeout_ms": 5000,
|
|
"optional": True,
|
|
},
|
|
{
|
|
"type": "evaluate",
|
|
"script": (
|
|
"(arg) => {"
|
|
"const setValue = (selector, value) => {"
|
|
"const el = document.querySelector(selector);"
|
|
"if (!el) return false;"
|
|
"el.value = value;"
|
|
"el.setAttribute('value', value);"
|
|
"el.dispatchEvent(new Event('input', {bubbles: true}));"
|
|
"el.dispatchEvent(new Event('change', {bubbles: true}));"
|
|
"return true;"
|
|
"};"
|
|
"return {"
|
|
"start: setValue('#startTime', arg.start),"
|
|
"end: setValue('#endTime', arg.end),"
|
|
"capital: setValue('#daily_backtest_capital_base_box', arg.capital)"
|
|
"};"
|
|
"}"
|
|
),
|
|
"arg": {
|
|
"start": "{backtest_start_date}",
|
|
"end": "{backtest_end_date}",
|
|
"capital": "{booksize}",
|
|
},
|
|
"description": "Set JoinQuant backtest dates and starting capital.",
|
|
"optional": True,
|
|
},
|
|
{
|
|
"type": "click",
|
|
"selector": "#algo-save-button",
|
|
"timeout_ms": 5000,
|
|
"description": "Save the JoinQuant strategy code.",
|
|
"optional": True,
|
|
},
|
|
{
|
|
"type": "click",
|
|
"selector": ".bootstrap-dialog .btn-primary, .modal.in button:has-text(\"确定\")",
|
|
"timeout_ms": 5000,
|
|
"force": True,
|
|
"description": "Confirm any JoinQuant save dialog.",
|
|
"optional": True,
|
|
},
|
|
{"type": "wait_for_timeout", "timeout_ms": 2000},
|
|
{
|
|
"type": "click",
|
|
"selector": "#daily-new-backtest-button",
|
|
"description": "Start the JoinQuant backtest.",
|
|
},
|
|
{
|
|
"type": "wait_for_selector",
|
|
"selector": "text=/回测完成|运行完成|Backtest complete|Finished/i",
|
|
"timeout_ms": 600_000,
|
|
"optional": True,
|
|
},
|
|
{
|
|
"type": "download",
|
|
"selector": "text=/导出成交|下载成交|fills|trades/i",
|
|
"save_as": "{expected_joinquant_csvs.fills}",
|
|
"timeout_ms": 15000,
|
|
"optional": True,
|
|
},
|
|
{
|
|
"type": "download",
|
|
"selector": "text=/导出持仓|下载持仓|positions/i",
|
|
"save_as": "{expected_joinquant_csvs.positions}",
|
|
"timeout_ms": 15000,
|
|
"optional": True,
|
|
},
|
|
{
|
|
"type": "download",
|
|
"selector": "text=/导出收益|下载收益|pnl|收益/i",
|
|
"save_as": "{expected_joinquant_csvs.pnl}",
|
|
"timeout_ms": 15000,
|
|
"optional": True,
|
|
},
|
|
{"type": "screenshot", "path": "{run_artifact_dir}/final.png"},
|
|
]
|
|
|
|
|
|
def _sim_trade_actions() -> list[dict[str, Any]]:
|
|
return [
|
|
{"type": "goto", "url": "{strategy_url}"},
|
|
{
|
|
"type": "paste_text_file",
|
|
"selector": "textarea, .ace_text-input, .cm-content, .CodeMirror textarea",
|
|
"path": "{wrapper_path}",
|
|
"description": "Paste generated wrapper strategy into the simulated-trading strategy editor.",
|
|
},
|
|
{
|
|
"type": "set_input_files",
|
|
"selector": "input[type=file]",
|
|
"paths": "{target_csvs}",
|
|
"description": "Upload frozen target CSV files for 模拟盘.",
|
|
"optional": True,
|
|
},
|
|
{
|
|
"type": "click",
|
|
"selector": "text=/保存|Save/i",
|
|
"description": "Save the strategy code and uploaded files.",
|
|
"optional": True,
|
|
},
|
|
{
|
|
"type": "wait_for_selector",
|
|
"selector": "text=/保存成功|已保存|Saved/i",
|
|
"timeout_ms": 120_000,
|
|
"optional": True,
|
|
},
|
|
{
|
|
"type": "click",
|
|
"selector": "text=/模拟盘|模拟交易|启动模拟|运行模拟|启动|重启|Run Sim|Start|Restart/i",
|
|
"description": "Start or restart the JoinQuant simulated-trading job.",
|
|
},
|
|
{
|
|
"type": "wait_for_selector",
|
|
"selector": "text=/运行中|已启动|模拟交易运行|Started|Running/i",
|
|
"timeout_ms": 180_000,
|
|
"optional": True,
|
|
},
|
|
{"type": "screenshot", "path": "{run_artifact_dir}/sim_trade_final.png"},
|
|
]
|
|
|
|
|
|
def default_browser_config(strategy_url: str = "", *, flow: str = "backtest") -> dict[str, Any]:
|
|
"""Return a selector/action template for JoinQuant browser automation."""
|
|
if flow not in {"backtest", "sim-trade", "sim_trade"}:
|
|
raise ValueError("flow must be 'backtest' or 'sim-trade'")
|
|
normalized_flow = "sim-trade" if flow == "sim_trade" else flow
|
|
actions = _sim_trade_actions() if normalized_flow == "sim-trade" else _backtest_actions()
|
|
return {
|
|
"flow": normalized_flow,
|
|
"strategy_url": strategy_url,
|
|
"login_url": DEFAULT_LOGIN_URL,
|
|
"headless": False,
|
|
"timeout_ms": 120_000,
|
|
"notes": [
|
|
"Fill strategy_url and selectors after inspecting your JoinQuant strategy page.",
|
|
"Use `joinquant browser-snapshot` to capture HTML/screenshots for selector tuning.",
|
|
"The action list is declarative and runs in order.",
|
|
],
|
|
"actions": actions,
|
|
}
|
|
|
|
|
|
def write_browser_config_template(
|
|
path: str | Path,
|
|
*,
|
|
strategy_url: str = "",
|
|
flow: str = "backtest",
|
|
) -> Path:
|
|
"""Write a JSON config template for browser automation."""
|
|
out_path = Path(path)
|
|
out_path.parent.mkdir(parents=True, exist_ok=True)
|
|
out_path.write_text(
|
|
json.dumps(
|
|
default_browser_config(strategy_url, flow=flow),
|
|
indent=2,
|
|
ensure_ascii=False,
|
|
) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
return out_path
|
|
|
|
|
|
def load_json(path: str | Path) -> dict[str, Any]:
|
|
"""Load a JSON object from disk."""
|
|
data = json.loads(Path(path).read_text(encoding="utf-8"))
|
|
if not isinstance(data, dict):
|
|
raise ValueError(f"Expected JSON object in {path}")
|
|
return data
|
|
|
|
|
|
def load_env_file(path: str | Path) -> dict[str, str]:
|
|
"""Load simple KEY=VALUE env files without requiring shell-safe quoting."""
|
|
values: dict[str, str] = {}
|
|
for raw in Path(path).expanduser().read_text(encoding="utf-8").splitlines():
|
|
line = raw.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
clean = value.strip()
|
|
if len(clean) >= 2 and clean[0] == clean[-1] and clean[0] in {"'", '"'}:
|
|
clean = clean[1:-1]
|
|
else:
|
|
clean = clean.strip("'\"")
|
|
values[key.strip()] = clean
|
|
return values
|
|
|
|
|
|
def _get_dotted(data: dict[str, Any], dotted: str) -> Any:
|
|
current: Any = data
|
|
for part in dotted.split("."):
|
|
if isinstance(current, dict) and part in current:
|
|
current = current[part]
|
|
else:
|
|
raise KeyError(dotted)
|
|
return current
|
|
|
|
|
|
def _manifest_context(manifest: dict[str, Any], artifact_dir: Path, config: dict[str, Any]) -> dict[str, Any]:
|
|
targets_dir = Path(str(manifest["targets_dir"]))
|
|
target_csvs = [str(path) for path in sorted(targets_dir.glob("*.csv"))]
|
|
if not target_csvs:
|
|
raise ValueError(f"No target CSV files found under {targets_dir}")
|
|
|
|
target_dates = [
|
|
datetime.strptime(Path(path).stem, "%Y%m%d").strftime("%Y-%m-%d")
|
|
for path in target_csvs
|
|
]
|
|
context = dict(manifest)
|
|
context.update({
|
|
"strategy_url": config.get("strategy_url", ""),
|
|
"target_csvs": target_csvs,
|
|
"target_csvs_csv": ",".join(target_csvs),
|
|
"backtest_start_date": config.get("backtest_start_date") or min(target_dates),
|
|
"backtest_end_date": config.get("backtest_end_date") or max(target_dates),
|
|
"run_artifact_dir": str(artifact_dir),
|
|
})
|
|
return context
|
|
|
|
|
|
_TOKEN_RE = re.compile(r"^\{([A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*)\}$")
|
|
_PARTIAL_TOKEN_RE = re.compile(r"\{([A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*)\}")
|
|
|
|
|
|
def resolve_template(value: Any, context: dict[str, Any]) -> Any:
|
|
"""Resolve ``{tokens}`` in config values against the manifest context."""
|
|
if isinstance(value, str):
|
|
whole = _TOKEN_RE.match(value)
|
|
if whole:
|
|
return _get_dotted(context, whole.group(1))
|
|
|
|
def replace(match: re.Match[str]) -> str:
|
|
resolved = _get_dotted(context, match.group(1))
|
|
return str(resolved)
|
|
|
|
return _PARTIAL_TOKEN_RE.sub(replace, value)
|
|
if isinstance(value, list):
|
|
return [resolve_template(item, context) for item in value]
|
|
if isinstance(value, dict):
|
|
return {key: resolve_template(val, context) for key, val in value.items()}
|
|
return value
|
|
|
|
|
|
def save_login_state(
|
|
*,
|
|
storage_state: str | Path,
|
|
login_url: str = DEFAULT_LOGIN_URL,
|
|
headless: bool = False,
|
|
wait_seconds: int = 0,
|
|
) -> Path:
|
|
"""Open a browser for manual login and save the authenticated state."""
|
|
if headless and wait_seconds <= 0:
|
|
raise ValueError("headless browser-login requires --wait-seconds > 0")
|
|
sync_playwright, _ = _require_playwright()
|
|
state_path = Path(storage_state).expanduser()
|
|
state_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with sync_playwright() as pw:
|
|
browser = pw.chromium.launch(headless=headless)
|
|
context = browser.new_context()
|
|
page = context.new_page()
|
|
page.goto(login_url, wait_until="domcontentloaded")
|
|
if wait_seconds > 0:
|
|
page.wait_for_timeout(wait_seconds * 1000)
|
|
else:
|
|
input("Log in to JoinQuant in the opened browser, then press Enter here to save state...")
|
|
context.storage_state(path=str(state_path))
|
|
browser.close()
|
|
state_path.chmod(0o600)
|
|
return state_path
|
|
|
|
|
|
def save_login_state_from_env(
|
|
*,
|
|
env_path: str | Path,
|
|
storage_state: str | Path,
|
|
login_url: str = DEFAULT_LOGIN_URL,
|
|
headless: bool = True,
|
|
out_dir: str | Path | None = None,
|
|
timeout_ms: int = 120_000,
|
|
) -> dict[str, Any]:
|
|
"""Try to log in with env-file credentials and save browser state.
|
|
|
|
This handles the normal password-login form. If JoinQuant presents CAPTCHA,
|
|
slide verification, SMS verification, or 2FA, the report will mark the run
|
|
as not logged in and save a screenshot for manual diagnosis.
|
|
"""
|
|
env = load_env_file(env_path)
|
|
username = env.get("JOINQUANT_USERNAME")
|
|
password = env.get("JOINQUANT_PASSWORD")
|
|
if not username or not password:
|
|
raise ValueError("JOINQUANT_USERNAME and JOINQUANT_PASSWORD are required")
|
|
|
|
sync_playwright, _ = _require_playwright()
|
|
state_path = Path(storage_state).expanduser()
|
|
state_path.parent.mkdir(parents=True, exist_ok=True)
|
|
artifact_dir = Path(out_dir or state_path.parent / "joinquant_login_artifacts")
|
|
artifact_dir.mkdir(parents=True, exist_ok=True)
|
|
screenshot_path = artifact_dir / "login_after_submit.png"
|
|
html_path = artifact_dir / "login_after_submit.html"
|
|
|
|
with sync_playwright() as pw:
|
|
browser = pw.chromium.launch(headless=headless)
|
|
context = browser.new_context()
|
|
page = context.new_page()
|
|
page.set_default_timeout(timeout_ms)
|
|
page.goto(login_url, wait_until="networkidle", timeout=timeout_ms)
|
|
page.locator('input[name="username"], input.pwd-phone').first.fill(username)
|
|
page.locator('input[name="pwd"], input.jq-login__password').first.fill(password)
|
|
checkbox = page.locator("#agreementBox, input.agreement-box").first
|
|
if checkbox.count():
|
|
checkbox.check(force=True)
|
|
page.locator("button.btnPwdSubmit, button.login-submit").first.click()
|
|
page.wait_for_timeout(5000)
|
|
|
|
html = page.content()
|
|
html_path.write_text(html, encoding="utf-8")
|
|
page.screenshot(path=str(screenshot_path), full_page=True)
|
|
login_inputs = page.locator('input[name="username"], input[name="pwd"]').count()
|
|
current_url = page.url
|
|
logged_in = login_inputs == 0 and "/user/login" not in current_url
|
|
if logged_in:
|
|
context.storage_state(path=str(state_path))
|
|
state_path.chmod(0o600)
|
|
browser.close()
|
|
|
|
report = {
|
|
"created_at": datetime.now(timezone.utc).isoformat(),
|
|
"logged_in": bool(logged_in),
|
|
"storage_state": str(state_path) if logged_in else "",
|
|
"artifact_dir": str(artifact_dir),
|
|
"screenshot": str(screenshot_path),
|
|
"html": str(html_path),
|
|
"current_url": current_url,
|
|
"notes": (
|
|
"Logged in and saved browser state."
|
|
if logged_in
|
|
else "Login did not complete. CAPTCHA/SMS/2FA or invalid credentials may be required."
|
|
),
|
|
}
|
|
report_path = artifact_dir / "login_report.json"
|
|
report["report_path"] = str(report_path)
|
|
report_path.write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
|
|
return report
|
|
|
|
|
|
def browser_snapshot(
|
|
*,
|
|
url: str,
|
|
storage_state: str | Path,
|
|
out_dir: str | Path,
|
|
headless: bool = True,
|
|
timeout_ms: int = 60_000,
|
|
) -> dict[str, Path]:
|
|
"""Save a logged-in page screenshot and HTML for selector discovery."""
|
|
sync_playwright, _ = _require_playwright()
|
|
root = Path(out_dir)
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
html_path = root / "page.html"
|
|
screenshot_path = root / "page.png"
|
|
|
|
with sync_playwright() as pw:
|
|
browser = pw.chromium.launch(headless=headless)
|
|
context = browser.new_context(storage_state=str(Path(storage_state).expanduser()))
|
|
page = context.new_page()
|
|
page.set_default_timeout(timeout_ms)
|
|
page.goto(url, wait_until="networkidle")
|
|
html_path.write_text(page.content(), encoding="utf-8")
|
|
page.screenshot(path=str(screenshot_path), full_page=True)
|
|
browser.close()
|
|
return {"html": html_path, "screenshot": screenshot_path}
|
|
|
|
|
|
def _locator(page: Any, selector: str):
|
|
return page.locator(selector).first
|
|
|
|
|
|
def _action_fail(action: dict[str, Any], exc: Exception) -> None:
|
|
if action.get("optional", False):
|
|
return
|
|
raise exc
|
|
|
|
|
|
def _run_action(page: Any, action: dict[str, Any], context: dict[str, Any], timeout_default: int) -> dict[str, Any]:
|
|
resolved = resolve_template(action, context)
|
|
kind = resolved["type"]
|
|
timeout_ms = int(resolved.get("timeout_ms", timeout_default))
|
|
record: dict[str, Any] = {"type": kind, "status": "ok"}
|
|
try:
|
|
if kind == "goto":
|
|
page.goto(str(resolved["url"]), wait_until=resolved.get("wait_until", "domcontentloaded"), timeout=timeout_ms)
|
|
elif kind == "click":
|
|
_locator(page, str(resolved["selector"])).click(
|
|
timeout=timeout_ms,
|
|
force=bool(resolved.get("force", False)),
|
|
)
|
|
elif kind == "fill":
|
|
_locator(page, str(resolved["selector"])).fill(str(resolved.get("text", "")), timeout=timeout_ms)
|
|
elif kind == "press":
|
|
_locator(page, str(resolved["selector"])).press(str(resolved["key"]), timeout=timeout_ms)
|
|
elif kind == "set_input_files":
|
|
paths = resolved.get("paths", [])
|
|
if isinstance(paths, str):
|
|
paths = [path for path in paths.split(",") if path]
|
|
_locator(page, str(resolved["selector"])).set_input_files(paths, timeout=timeout_ms)
|
|
record["n_files"] = len(paths)
|
|
elif kind == "paste_text_file":
|
|
text = Path(str(resolved["path"])).read_text(encoding="utf-8")
|
|
loc = _locator(page, str(resolved["selector"]))
|
|
loc.click(timeout=timeout_ms)
|
|
page.keyboard.press("Control+A")
|
|
page.keyboard.insert_text(text)
|
|
record["n_chars"] = len(text)
|
|
elif kind == "set_ace_editor_file":
|
|
text = Path(str(resolved["path"])).read_text(encoding="utf-8")
|
|
selector = str(resolved.get("selector", ".ace_editor"))
|
|
page.wait_for_function(
|
|
"(selector) => Boolean(window.ace && document.querySelector(selector))",
|
|
arg=selector,
|
|
timeout=timeout_ms,
|
|
)
|
|
page.evaluate(
|
|
"""(arg) => {
|
|
const node = document.querySelector(arg.selector);
|
|
if (!node || !window.ace) {
|
|
throw new Error(`Ace editor not found for ${arg.selector}`);
|
|
}
|
|
const editor = window.ace.edit(node);
|
|
editor.setValue(arg.text, -1);
|
|
editor.clearSelection();
|
|
editor.focus();
|
|
return editor.getValue().length;
|
|
}""",
|
|
{"selector": selector, "text": text},
|
|
)
|
|
record["n_chars"] = len(text)
|
|
elif kind == "evaluate":
|
|
page.evaluate(str(resolved["script"]), resolved.get("arg"))
|
|
elif kind == "wait_for_selector":
|
|
page.wait_for_selector(str(resolved["selector"]), timeout=timeout_ms)
|
|
elif kind == "wait_for_timeout":
|
|
page.wait_for_timeout(int(resolved.get("timeout_ms", 1000)))
|
|
elif kind == "download":
|
|
save_as = Path(str(resolved["save_as"]))
|
|
save_as.parent.mkdir(parents=True, exist_ok=True)
|
|
with page.expect_download(timeout=timeout_ms) as download_info:
|
|
_locator(page, str(resolved["selector"])).click(timeout=timeout_ms)
|
|
download = download_info.value
|
|
download.save_as(str(save_as))
|
|
record["path"] = str(save_as)
|
|
elif kind == "screenshot":
|
|
path = Path(str(resolved["path"]))
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
page.screenshot(path=str(path), full_page=bool(resolved.get("full_page", True)))
|
|
record["path"] = str(path)
|
|
else:
|
|
raise ValueError(f"Unsupported browser action type: {kind}")
|
|
except Exception as exc: # pragma: no cover - exercised only with Playwright.
|
|
record["status"] = "skipped" if resolved.get("optional", False) else "failed"
|
|
record["error"] = str(exc)
|
|
_action_fail(resolved, exc)
|
|
return record
|
|
|
|
|
|
def run_browser_flow(
|
|
*,
|
|
manifest_path: str | Path,
|
|
config_path: str | Path,
|
|
storage_state: str | Path,
|
|
out_dir: str | Path | None = None,
|
|
headless: bool | None = None,
|
|
auto_reconcile: bool = True,
|
|
flow_name: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Run configured browser automation for a JoinQuant web workflow."""
|
|
sync_playwright, _ = _require_playwright()
|
|
manifest = load_json(manifest_path)
|
|
config = load_json(config_path)
|
|
flow = flow_name or str(config.get("flow") or "browser")
|
|
artifact_dir = Path(out_dir or Path(str(manifest["joinquant_export_dir"])).parent / f"browser_{flow}")
|
|
artifact_dir.mkdir(parents=True, exist_ok=True)
|
|
context_vars = _manifest_context(manifest, artifact_dir, config)
|
|
|
|
timeout_ms = int(config.get("timeout_ms", 120_000))
|
|
actions = config.get("actions") or []
|
|
if not actions:
|
|
raise ValueError("Browser config contains no actions")
|
|
|
|
records: list[dict[str, Any]] = []
|
|
failure: Exception | None = None
|
|
failure_artifacts: dict[str, str] = {}
|
|
with sync_playwright() as pw:
|
|
browser = pw.chromium.launch(
|
|
headless=bool(config.get("headless", False) if headless is None else headless)
|
|
)
|
|
context = browser.new_context(
|
|
storage_state=str(Path(storage_state).expanduser()),
|
|
accept_downloads=True,
|
|
)
|
|
page = context.new_page()
|
|
page.set_default_timeout(timeout_ms)
|
|
try:
|
|
for action in actions:
|
|
records.append(_run_action(page, action, context_vars, timeout_ms))
|
|
except Exception as exc: # pragma: no cover - requires live browser UI.
|
|
failure = exc
|
|
screenshot_path = artifact_dir / "failure.png"
|
|
html_path = artifact_dir / "failure.html"
|
|
try:
|
|
page.screenshot(path=str(screenshot_path), full_page=True)
|
|
html_path.write_text(page.content(), encoding="utf-8")
|
|
failure_artifacts = {
|
|
"screenshot": str(screenshot_path),
|
|
"html": str(html_path),
|
|
}
|
|
except Exception as artifact_exc:
|
|
failure_artifacts = {"artifact_error": str(artifact_exc)}
|
|
records.append({
|
|
"type": "browser_flow",
|
|
"status": "failed",
|
|
"error": str(exc),
|
|
**failure_artifacts,
|
|
})
|
|
browser.close()
|
|
|
|
expected = manifest.get("expected_joinquant_csvs", {})
|
|
downloaded = {
|
|
key: str(path)
|
|
for key, value in expected.items()
|
|
if (path := Path(str(value))).exists()
|
|
}
|
|
reconcile_paths: dict[str, str] = {}
|
|
if auto_reconcile and {"fills", "positions", "pnl"}.issubset(downloaded):
|
|
ingested = ingest_joinquant_outputs(
|
|
portfolio_name=str(manifest["portfolio_name"]),
|
|
fills_csv=downloaded["fills"],
|
|
positions_csv=downloaded["positions"],
|
|
pnl_csv=downloaded["pnl"],
|
|
out_dir=Path(str(manifest["joinquant_export_dir"])).parent / "ingested",
|
|
)
|
|
reconciled = reconcile_joinquant(
|
|
portfolio_name=str(manifest["portfolio_name"]),
|
|
targets_dir=str(manifest["targets_dir"]),
|
|
our_fills_path=str(manifest["fills_path"]),
|
|
our_positions_path=str(manifest["positions_path"]),
|
|
our_pnl_path=str(manifest["pnl_path"]),
|
|
jq_fills_path=str(ingested["fills"]),
|
|
jq_positions_path=str(ingested["positions"]),
|
|
jq_pnl_path=str(ingested["pnl"]),
|
|
out_dir=Path(str(manifest["joinquant_export_dir"])).parent / "reconcile",
|
|
)
|
|
reconcile_paths = {key: str(value) for key, value in reconciled.items()}
|
|
|
|
report = {
|
|
"created_at": datetime.now(timezone.utc).isoformat(),
|
|
"manifest_path": str(manifest_path),
|
|
"config_path": str(config_path),
|
|
"flow": flow,
|
|
"status": "failed" if failure else "ok",
|
|
"storage_state": str(storage_state),
|
|
"artifact_dir": str(artifact_dir),
|
|
"actions": records,
|
|
"downloaded": downloaded,
|
|
"reconcile_paths": reconcile_paths,
|
|
}
|
|
report_path = artifact_dir / "browser_run_report.json"
|
|
report["report_path"] = str(report_path)
|
|
report_path.write_text(
|
|
json.dumps(report, indent=2, ensure_ascii=False) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
if failure is not None:
|
|
raise RuntimeError(
|
|
f"Browser flow failed; report saved to {report_path}: {failure}"
|
|
) from failure
|
|
return report
|
|
|
|
|
|
def run_browser_backtest(
|
|
*,
|
|
manifest_path: str | Path,
|
|
config_path: str | Path,
|
|
storage_state: str | Path,
|
|
out_dir: str | Path | None = None,
|
|
headless: bool | None = None,
|
|
auto_reconcile: bool = True,
|
|
) -> dict[str, Any]:
|
|
"""Run configured browser automation for a JoinQuant backtest."""
|
|
return run_browser_flow(
|
|
manifest_path=manifest_path,
|
|
config_path=config_path,
|
|
storage_state=storage_state,
|
|
out_dir=out_dir,
|
|
headless=headless,
|
|
auto_reconcile=auto_reconcile,
|
|
flow_name="backtest",
|
|
)
|
|
|
|
|
|
def run_browser_sim_trade(
|
|
*,
|
|
manifest_path: str | Path,
|
|
config_path: str | Path,
|
|
storage_state: str | Path,
|
|
out_dir: str | Path | None = None,
|
|
headless: bool | None = None,
|
|
auto_reconcile: bool = True,
|
|
) -> dict[str, Any]:
|
|
"""Run configured browser automation for JoinQuant 模拟盘."""
|
|
return run_browser_flow(
|
|
manifest_path=manifest_path,
|
|
config_path=config_path,
|
|
storage_state=storage_state,
|
|
out_dir=out_dir,
|
|
headless=headless,
|
|
auto_reconcile=auto_reconcile,
|
|
flow_name="sim-trade",
|
|
)
|