Hermes-Skills/longbridge-cli/scripts/longbridge_cli_helper.py

#!/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])