#!/usr/bin/env python3 """ longbridge_cli_helper.py - SDK 兼容层, 内部走 CLI 提供给日内监控脚本用, 避免 Python SDK 的 602315 问题 环境变量要求: LONGBRIDGE_HTTP_URL=https://openapi.longbridge.com LONGBRIDGE_REGION=ap LONGBRIDGE_TRADE_ENABLED=true LONGBRIDGE_* / LONGPORT_* 在 ~/.bashrc 每个函数调用都包 proxychains4 """ import os import subprocess import shlex import re import json PROXY = 'proxychains4 -f ~/.proxychains/proxychains.conf' CLI = '/home/openclaw/.local/bin/longbridge' PROFILE = 'lb_real' def _run_longbridge(*args, env_extra=None): """执行 longbridge CLI 命令, 返回 stdout""" env = os.environ.copy() # 强制 .com 海外域 (避免 602315) env['LONGBRIDGE_HTTP_URL'] = 'https://openapi.longbridge.com' env['LONGBRIDGE_REGION'] = 'ap' env['LONGBRIDGE_TRADE_ENABLED'] = 'true' # 加载 LONGPORT_* 凭证 (CLI 也读) bashrc = open(os.path.expanduser('~/.bashrc')).read() for line in bashrc.splitlines(): if line.startswith('export LONGPORT_') or line.startswith('export LONGBRIDGE_'): parts = line.replace('export ', '').split('=', 1) if len(parts) == 2: env[parts[0]] = parts[1].strip('"').strip("'") if env_extra: env.update(env_extra) cmd = f"{PROXY} {CLI} --profile {PROFILE} " + ' '.join(shlex.quote(str(a)) for a in args) result = subprocess.run(cmd, shell=True, capture_output=True, text=True, env=env, timeout=30) if result.returncode != 0: raise RuntimeError(f"CLI error: {result.stderr.strip()}") return result.stdout def _parse_table(text): """CLI 输出是表格, 转成 list of dict 注意: header 用 ┃ (U+2503), data 用 │ (U+2502) """ lines = text.split('\n') table_lines = [l for l in lines if '┃' in l or '│' in l] if len(table_lines) < 2: return [] def split_row(line): cells = re.split('[┃│]', line) return [c.strip() for c in cells if c.strip()] headers = split_row(table_lines[0]) result = [] for row in table_lines[1:]: cols = split_row(row) if not cols: continue try: d = {h: cols[i] if i < len(cols) else '' for i, h in enumerate(headers)} result.append(d) except IndexError: continue return result class AccountBalance: def __init__(self, buy_power, currency='HKD', total_cash=0, net_assets=0): self.buy_power = buy_power self.currency = currency self.total_cash = total_cash self.net_assets = net_assets def account_balance(): """获取账户余额 (CLI 没有 buy_power, 用 现金 + 剩余融资 推算)""" text = _run_longbridge('balance') rows = _parse_table(text) if not rows: return [AccountBalance(buy_power=0)] r = rows[0] try: cash = float(r.get('现金余额', '0').replace(',', '')) finance = float(r.get('剩余融资额', '0').replace(',', '')) net = float(r.get('净资产', '0').replace(',', '')) bp = cash + finance currency = r.get('币种', 'HKD').strip() return [AccountBalance(buy_power=bp, currency=currency, total_cash=cash, net_assets=net)] except (ValueError, KeyError) as e: return [AccountBalance(buy_power=0)] class Position: def __init__(self, symbol, quantity, cost_price, available_quantity=None): self.symbol = symbol self.quantity = quantity self.cost_price = cost_price self.available_quantity = available_quantity or quantity def stock_positions(): """获取持仓, 返回类似 SDK 的结构""" text = _run_longbridge('positions') rows = _parse_table(text) class Channels: def __init__(self, positions): self.channels = [type('C', (), {'positions': positions})()] positions = [] for r in rows: sym = r.get('标的', '').strip() qty_str = r.get('持仓', '0').strip().replace(',', '') if not sym or not qty_str or not qty_str.isdigit(): continue try: qty = int(qty_str) cost = float(r.get('成本价', '0').replace(',', '')) avail = int(r.get('可卖数量', str(qty)).replace(',', '')) if qty > 0: positions.append(Position(sym, qty, cost, avail)) except (ValueError, KeyError): continue return Channels(positions) class OrderResult: def __init__(self, order_id): self.order_id = order_id def submit_order(symbol, order_type, side, submitted_quantity, time_in_force, submitted_price=None, **kwargs): """下单 - CLI 包装""" side_str = 'buy' if str(side).endswith('Buy') else 'sell' if str(order_type).endswith('MO'): args = ['sell' if side_str == 'sell' else 'buy', symbol, '--qty', submitted_quantity, '-y'] if submitted_price: args.extend(['--price', submitted_price]) else: args = [side_str, symbol, '--qty', submitted_quantity, '--price', submitted_price, '-y'] text = _run_longbridge(*args) match = re.search(r'订单号[：:]\s*(\d+)', text) if match: return OrderResult(match.group(1)) raise RuntimeError(f"下单失败: {text.strip()}") def cancel_order(order_id): """撤单 - CLI 强制 y (cancel 没有 -y)""" env = os.environ.copy() env['LONGBRIDGE_HTTP_URL'] = 'https://openapi.longbridge.com' env['LONGBRIDGE_REGION'] = 'ap' env['LONGBRIDGE_TRADE_ENABLED'] = 'true' bashrc = open(os.path.expanduser('~/.bashrc')).read() for line in bashrc.splitlines(): if line.startswith('export LONGPORT_') or line.startswith('export LONGBRIDGE_'): parts = line.replace('export ', '').split('=', 1) if len(parts) == 2: env[parts[0]] = parts[1].strip('"').strip("'") cmd = f"echo 'y' | {PROXY} {CLI} --profile {PROFILE} cancel {shlex.quote(str(order_id))}" subprocess.run(cmd, shell=True, env=env, timeout=30) # enums 兼容 class OrderType: LO = 'LO' MO = 'MO' ELO = 'ELO' class OrderSide: Buy = 'Buy' Sell = 'Sell' class TimeInForceType: Day = 'Day' GoodTilCanceled = 'GoodTilCanceled' # 测试 if __name__ == '__main__': print("=== balance ===") bals = account_balance() for b in bals: print(f"buy_power: {b.buy_power}") print("\n=== positions ===") pos = stock_positions() for ch in pos.channels: for p in ch.positions: print(f"{p.symbol}: {p.quantity}股 @ {p.cost_price}") print("\n=== orders ===") text = _run_longbridge('orders') print(text[:500])