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

#!/usr/bin/env python3
"""
通用持仓做T工具 - 不限定股票,根据命令行参数查任意持仓
用法:
  python3 stock_t.py RGTI.US status      - 查看某股票持仓/挂单
  python3 stock_t.py RGTI.US plan        - 查看做T计划（不执行）
  python3 stock_t.py RGTI.US execute     - 半自动执行（需确认）
  python3 stock_t.py RGTI.US auto        - 全自动执行（直接挂单）
  python3 stock_t.py RGTI.US cancel      - 撤销某股票所有挂单
  python3 stock_t.py list                - 列出所有持仓

Requires 602315 bypass to actually trade:
  proxychains4 -f ~/.proxychains/proxychains.conf python3 stock_t.py <args>
"""
import os, sys, json

os.environ['LONGBRIDGE_REGION'] = 'ap'

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:
            os.environ[parts[0]] = parts[1].strip('"').strip("'")

from longport import openapi

cfg = openapi.Config.from_env()
trade_ctx = openapi.TradeContext(config=cfg)
quote_ctx = openapi.QuoteContext(config=cfg)

if len(sys.argv) < 2:
    print(__doc__)
    sys.exit(1)

KNOWN_COMMANDS = {'list', 'status', 'plan', 'execute', 'auto', 'cancel'}

if sys.argv[1].lower() in KNOWN_COMMANDS:
    cmd = sys.argv[1].lower()
    if cmd != 'list' and len(sys.argv) < 3:
        print("错误: 需要股票代码,例如 RGTI.US")
        sys.exit(1)
    SYMBOL = sys.argv[2].upper() if cmd != 'list' and len(sys.argv) > 2 else None
else:
    if len(sys.argv) < 3:
        print("错误: 用法: stock_t.py <SYMBOL> <command> 或 stock_t.py list")
        sys.exit(1)
    SYMBOL = sys.argv[1].upper()
    cmd = sys.argv[2].lower()
    if cmd not in KNOWN_COMMANDS:
        print(f"未知命令: {cmd}")
        sys.exit(1)

if cmd == 'list':
    print("=== 长桥全部持仓 ===")
    positions = trade_ctx.stock_positions()
    total_value = 0
    for ch in positions.channels:
        for p in ch.positions:
            try:
                cost = float(p.cost_price)
                qty = int(p.quantity)
                val = cost * qty
                total_value += val
                avail = int(getattr(p, 'available_quantity', qty))
                print(f"  {p.symbol}: {qty}股 @ ${cost:.2f} = ${val:.2f} (可卖:{avail})")
            except Exception as e:
                print(f"  {p.symbol}: 解析失败 {e}")
    print(f"\n持仓总市值: ${total_value:.2f}")
    sys.exit(0)

CONFIG_FILE = os.path.expanduser(f"~/.hermes/scripts/{SYMBOL.replace('.', '_').lower()}_t_config.json")
T_CONFIG = {
    "symbol": SYMBOL,
    "trade_qty": None,
    "buy_levels": [],
    "sell_levels": [],
    "spread_buffer": 0.10,
}

if os.path.exists(CONFIG_FILE):
    try:
        custom = json.load(open(CONFIG_FILE))
        T_CONFIG.update(custom)
    except Exception:
        pass


def get_quote(symbol):
    q = quote_ctx.quote([symbol])[0]
    return float(q.last_done), float(q.high), float(q.low), float(q.prev_close)


def get_position(symbol):
    positions = trade_ctx.stock_positions()
    for ch in positions.channels:
        for p in ch.positions:
            if p.symbol == symbol:
                return p
    return None


def get_orders(symbol):
    orders = trade_ctx.today_orders()
    return [o for o in orders if o.symbol == symbol]


def cmd_status():
    pos = get_position(SYMBOL)
    price, high, low, prev = get_quote(SYMBOL)

    print(f"\n=== {SYMBOL} 实时行情 ===")
    print(f"现价: ${price:.2f}")
    print(f"日内高: ${high:.2f} | 日内低: ${low:.2f}")
    print(f"昨收: ${prev:.2f} | 涨跌: {(price-prev)/prev*100:+.2f}%")

    if pos:
        cost = float(pos.cost_price)
        qty = int(pos.quantity)
        avail = int(getattr(pos, 'available_quantity', qty))
        upl = (price - cost) * qty
        upl_pct = (price - cost) / cost * 100
        print(f"\n=== {SYMBOL} 持仓 ===")
        print(f"数量: {qty}股 (可卖:{avail})")
        print(f"成本: ${cost:.2f} | 现价: ${price:.2f}")
        print(f"浮盈: {upl:+.2f} USDT ({upl_pct:+.2f}%)")
    else:
        print(f"\n=== {SYMBOL} 无持仓 ===")

    orders = get_orders(SYMBOL)
    if orders:
        print(f"\n=== 今日挂单 ===")
        for o in orders:
            print(f"  {o.order_id} | {o.side} | {o.quantity}股 @ ${o.price} | {o.status}")
    else:
        print(f"\n无挂单")


def cmd_plan():
    pos = get_position(SYMBOL)
    if not pos:
        print(f"❌ {SYMBOL} 无持仓,无法做T")
        return

    qty = int(pos.quantity)
    cost = float(pos.cost_price)
    price, high, low, prev = get_quote(SYMBOL)

    print(f"\n=== {SYMBOL} 做T计划 ===")
    print(f"持仓: {qty}股 @ ${cost:.2f}")
    print(f"现价: ${price:.2f} (浮盈: {(price-cost)*qty:+.2f})")

    if not T_CONFIG['buy_levels'] or not T_CONFIG['sell_levels']:
        print(f"\n未配置 buy_levels / sell_levels")
        print(f"创建 {CONFIG_FILE}:")
        print(json.dumps({
            "trade_qty": qty,
            "buy_levels": [round(price*0.95, 2), round(price*0.90, 2), round(price*0.85, 2)],
            "sell_levels": [round(price*1.05, 2), round(price*1.10, 2), round(price*1.15, 2)],
            "spread_buffer": 0.10
        }, indent=2))
        return

    print(f"\n=== 买入触发位 ===")
    for lv in T_CONFIG['buy_levels']:
        print(f"  ${lv:.2f} (现价-{abs(price-lv):.2f})")

    print(f"\n=== 卖出触发位 ===")
    for lv in T_CONFIG['sell_levels']:
        print(f"  ${lv:.2f} (现价+{abs(price-lv):.2f})")


def cmd_cancel():
    orders = get_orders(SYMBOL)
    if not orders:
        print(f"{SYMBOL} 无挂单")
        return

    print(f"撤销 {SYMBOL} 的 {len(orders)} 个挂单:")
    for o in orders:
        print(f"  {o.order_id} | {o.side} | {o.quantity}股 @ ${o.price}")
        try:
            trade_ctx.cancel_order(o.order_id)
            print(f"    已撤")
        except Exception as e:
            print(f"    失败: {e}")


if cmd == 'status':
    cmd_status()
elif cmd == 'plan':
    cmd_plan()
elif cmd == 'execute':
    print(">>> 用 stock_t.py <SYMBOL> plan 查看计划,然后用 longbridge CLI 下单")
elif cmd == 'auto':
    print(">>> 手动下单: LONGBRIDGE_REGION=ap LONGBRIDGE_TRADE_ENABLED=true proxychains4 -f ~/.proxychains/proxychains.conf ~/.local/bin/longbridge --profile lb_real buy/sell <SYM> --qty N --price P -y")
elif cmd == 'cancel':
    cmd_cancel()
else:
    print(f"未知命令: {cmd}")
    sys.exit(1)