hello-agents/Co-creation-projects/lcyting-StockSage-agent/skills/模拟组合管理/mx-moni/mx_moni.py

#!/usr/bin/env python3
# mx_moni - 妙想模拟组合管理技能

import os
import sys
import json
import re
import requests
from typing import Dict, Any, Optional, Tuple

# 加载环境变量
MX_APIKEY = os.environ.get('MX_APIKEY')
MX_API_URL = os.environ.get('MX_API_URL', 'https://mkapi2.dfcfs.com/finskillshub')
OUTPUT_DIR = '/root/.openclaw/workspace/mx_data/output'

os.makedirs(OUTPUT_DIR, exist_ok=True)

def check_apikey() -> None:
    """检查API密钥是否配置"""
    if not MX_APIKEY:
        print("错误: 未配置MX_APIKEY环境变量，请先配置API密钥")
        print("示例: export MX_APIKEY=your_api_key_here")
        sys.exit(1)

def make_request(endpoint: str, body: Dict[str, Any], output_prefix: str) -> None:
    """发送POST请求并保存结果"""
    check_apikey()
    full_url = f"{MX_API_URL}{endpoint}"
    headers = {
        'apikey': MX_APIKEY,
        'Content-Type': 'application/json'
    }
    
    try:
        response = requests.post(full_url, headers=headers, json=body)
        response.raise_for_status()
        result = response.json()
        
        output_path = os.path.join(OUTPUT_DIR, f"{output_prefix}_raw.json")
        with open(output_path, 'w', encoding='utf-8') as f:
            json.dump(result, f, ensure_ascii=False, indent=2)
        
        print(f"请求完成，结果保存在 {output_path}")
        
        # 打印结果摘要
        if result.get('success') or str(result.get('code')) == '200':
            print("\n操作结果: 成功")
            if 'message' in result:
                print(f"提示信息: {result['message']}")
            if 'data' in result and isinstance(result['data'], dict):
                data = result['data']
                if 'totalAssets' in data:
                    print(f"\n账户资金:")
                    print(f"  总资产: {data['totalAssets']:.2f} 元")
                    print(f"  可用资金: {data['availBalance']:.2f} 元")
                if 'orderId' in data:
                    print(f"\n委托成功:")
                    print(f"  委托编号: {data['orderId']}")
        else:
            print(f"\n操作结果: 失败")
            print(f"错误码: {result.get('code')}")
            print(f"错误信息: {result.get('message')}")
    except Exception as e:
        print(f"网络请求失败: {str(e)}")
        sys.exit(1)

def parse_buy_sell(query: str) -> Tuple[Optional[str], Optional[float], Optional[int], bool]:
    """解析买入卖出命令，返回(股票代码, 价格, 数量, 是否市价)"""
    # 提取6位股票代码
    code_match = re.search(r'(\d{6})', query)
    if not code_match:
        return None, None, None, False
    stock_code = code_match.group(1)
    
    # 提取数量（单位：股，必须是100倍数）
    quantity_match = re.search(r'(\d+)\s*(股|手)', query)
    quantity = None
    if quantity_match:
        qty = int(quantity_match.group(1))
        if quantity_match.group(2) == '手':
            qty = qty * 100
        quantity = qty
    
    # 检查是否市价委托
    is_market = any(word in query for word in ['市价', '市价买入', '市价卖出', '现价买入', '现价卖出'])
    
    # 提取价格
    price_match = re.search(r'(\d+\.?\d*)\s*元', query) if not is_market else None
    price = None
    if price_match and not is_market:
        price = float(price_match.group(1))
    elif not is_market and quantity:
        # 尝试找任意数字作为价格
        price_candidates = re.findall(r'\d+\.?\d*', query)
        for candidate in price_candidates:
            if len(candidate) != 6:  # 排除股票代码
                price = float(candidate)
                break
    
    return stock_code, price, quantity, is_market

def parse_cancel(query: str) -> Tuple[Optional[str], Optional[str], bool]:
    """解析撤单命令，返回(委托编号, 股票代码, 是否全部撤单)"""
    if any(word in query for word in ['全部', '所有', '一键撤单']):
        return None, None, True
    
    # 提取委托编号
    order_id_match = re.search(r'(\d{16,20})', query)
    order_id = order_id_match.group(1) if order_id_match else None
    
    # 提取股票代码
    code_match = re.search(r'(\d{6})', query)
    stock_code = code_match.group(1) if code_match else None
    
    return order_id, stock_code, False

def main():
    if len(sys.argv) < 2:
        print("请提供操作指令，例如：")
        print("  python mx_moni.py 我的持仓      # 查询持仓")
        print("  python mx_moni.py 我的资金      # 查询资金")
        print("  python mx_moni.py 我的委托      # 查询委托订单")
        print("  python mx_moni.py 买入 600519 价格 1700 数量 100 股")
        print("  python mx_moni.py 市价买入 600519 100 股")
        print("  python mx_moni.py 卖出 600519 价格 1750 数量 100 股")
        print("  python mx_moni.py 撤单 123456789012345678")
        print("  python mx_moni.py 一键撤单")
        sys.exit(1)
    
    query = ' '.join(sys.argv[1:])
    output_prefix = f"mx_moni_{query.replace(' ', '_')}"
    
    # 根据意图识别调用不同接口
    if any(word in query for word in ['持仓', '我的持仓', '持仓情况']):
        make_request('/api/claw/mockTrading/positions', {'moneyUnit': 1}, output_prefix)
    elif any(word in query for word in ['资金', '我的资金', '账户余额', '资金情况']):
        make_request('/api/claw/mockTrading/balance', {'moneyUnit': 1}, output_prefix)
    elif any(word in query for word in ['委托', '我的委托', '订单', '委托记录']):
        make_request('/api/claw/mockTrading/orders', {'fltOrderDrt': 0, 'fltOrderStatus': 0}, output_prefix)
    elif any(word in query for word in ['买入', '买进', '建仓']):
        stock_code, price, quantity, is_market = parse_buy_sell(query)
        if not stock_code or not quantity:
            print("错误: 无法解析买入指令，请确保包含股票代码(6位)和数量(100的整数倍)")
            print("示例: python mx_moni.py 买入 600519 价格 1700 数量 100 股")
            print("示例: python mx_moni.py 市价买入 600519 100 股")
            sys.exit(1)
        if not is_market and price is None:
            print("错误: 限价买入需要提供价格，或使用市价买入")
            sys.exit(1)
        if quantity % 100 != 0:
            print("错误: 委托数量必须为100的整数倍")
            sys.exit(1)
        
        body = {
            'type': 'buy',
            'stockCode': stock_code,
            'quantity': quantity,
            'useMarketPrice': is_market
        }
        if not is_market:
            body['price'] = price
        
        make_request('/api/claw/mockTrading/trade', body, output_prefix)
    elif any(word in query for word in ['卖出', '抛售', '减仓']):
        stock_code, price, quantity, is_market = parse_buy_sell(query)
        if not stock_code or not quantity:
            print("错误: 无法解析卖出指令，请确保包含股票代码(6位)和数量(100的整数倍)")
            print("示例: python mx_moni.py 卖出 600519 价格 1750 数量 100 股")
            print("示例: python mx_moni.py 市价卖出 600519 100 股")
            sys.exit(1)
        if not is_market and price is None:
            print("错误: 限价卖出需要提供价格，或使用市价卖出")
            sys.exit(1)
        if quantity % 100 != 0:
            print("错误: 委托数量必须为100的整数倍")
            sys.exit(1)
        
        body = {
            'type': 'sell',
            'stockCode': stock_code,
            'quantity': quantity,
            'useMarketPrice': is_market
        }
        if not is_market:
            body['price'] = price
        
        make_request('/api/claw/mockTrading/trade', body, output_prefix)
    elif any(word in query for word in ['撤单', '撤销', '撤单']):
        order_id, stock_code, is_all = parse_cancel(query)
        if is_all:
            body = {'type': 'all'}
            make_request('/api/claw/mockTrading/cancel', body, output_prefix)
        else:
            if not order_id:
                print("错误: 请提供委托编号，或使用一键撤单撤销所有未成交委托")
                print("示例: python mx_moni.py 撤单 260854300000078983")
                print("示例: python mx_moni.py 一键撤单")
                sys.exit(1)
            body = {
                'type': 'order',
                'orderId': order_id
            }
            if stock_code:
                body['stockCode'] = stock_code
            make_request('/api/claw/mockTrading/cancel', body, output_prefix)
    else:
        print("无法识别意图，请使用以下操作之一：")
        print("  持仓查询: 我的持仓 / 查询持仓")
        print("  资金查询: 我的资金 / 查询资金")
        print("  委托查询: 我的委托 / 查询委托")
        print("  买入操作: 买入 [股票代码] [价格] [数量] 股 / 市价买入 [股票代码] [数量] 股")
        print("  卖出操作: 卖出 [股票代码] [价格] [数量] 股 / 市价卖出 [股票代码] [数量] 股")
        print("  撤单操作: 撤单 [委托编号] / 一键撤单")
        sys.exit(1)

if __name__ == '__main__':
    main()