127 lines
3.9 KiB
Python
127 lines
3.9 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
|
|
from typing import Dict, Any, List, Optional
|
|
from datetime import datetime
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import Column, Integer, String, DateTime, Index
|
|
|
|
from cryptoai.models.base import Base, logger
|
|
from cryptoai.utils.db_manager import get_db_context
|
|
# 定义Token模型
|
|
class Token(Base):
|
|
"""Token信息表模型"""
|
|
__tablename__ = 'tokens'
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
symbol = Column(String(50), nullable=False, unique=True, comment='交易对符号')
|
|
base_asset = Column(String(20), nullable=False, comment='基础资产')
|
|
quote_asset = Column(String(20), nullable=False, comment='计价资产')
|
|
created_at = Column(DateTime, nullable=False, default=datetime.now, comment='创建时间')
|
|
|
|
# 索引
|
|
__table_args__ = (
|
|
Index('idx_symbol', 'symbol'),
|
|
Index('idx_base_asset', 'base_asset'),
|
|
Index('idx_quote_asset', 'quote_asset'),
|
|
{'mysql_charset': 'utf8mb4', 'mysql_collate': 'utf8mb4_unicode_ci'}
|
|
)
|
|
|
|
class TokenManager:
|
|
"""Token管理类"""
|
|
|
|
def __init__(self, session: Session = None):
|
|
self.session = session
|
|
|
|
def create_token(self, symbol: str, base_asset: str, quote_asset: str) -> bool:
|
|
"""
|
|
创建新的Token信息
|
|
|
|
Args:
|
|
symbol: 交易对符号
|
|
base_asset: 基础资产
|
|
quote_asset: 计价资产
|
|
|
|
Returns:
|
|
创建是否成功
|
|
"""
|
|
try:
|
|
# 检查交易对是否已存在
|
|
existing_token = self.session.query(Token).filter(Token.symbol == symbol).first()
|
|
if existing_token:
|
|
logger.warning(f"交易对 {symbol} 已存在")
|
|
return False
|
|
|
|
# 创建新Token记录
|
|
new_token = Token(
|
|
symbol=symbol,
|
|
base_asset=base_asset,
|
|
quote_asset=quote_asset
|
|
)
|
|
|
|
self.session.add(new_token)
|
|
self.session.commit()
|
|
|
|
logger.info(f"成功创建Token信息: {symbol}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
self.session.rollback()
|
|
logger.error(f"创建Token信息失败: {e}")
|
|
return False
|
|
|
|
def delete_token(self, symbol: str) -> bool:
|
|
"""
|
|
删除Token信息
|
|
|
|
Args:
|
|
symbol: 交易对符号
|
|
|
|
Returns:
|
|
删除是否成功
|
|
"""
|
|
try:
|
|
# 查询Token
|
|
token = self.session.query(Token).filter(Token.symbol == symbol).first()
|
|
if not token:
|
|
logger.warning(f"交易对 {symbol} 不存在")
|
|
return False
|
|
|
|
# 删除Token
|
|
self.session.delete(token)
|
|
self.session.commit()
|
|
|
|
logger.info(f"成功删除Token信息: {symbol}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
self.session.rollback()
|
|
logger.error(f"删除Token信息失败: {e}")
|
|
return False
|
|
|
|
def search_token(self, key: str, limit: int = 10) -> List[Dict[str, Any]]:
|
|
"""
|
|
搜索Token
|
|
|
|
Args:
|
|
key: 搜索关键词
|
|
limit: 最大返回数量
|
|
|
|
Returns:
|
|
Token信息列表
|
|
"""
|
|
try:
|
|
# 使用 SQLAlchemy 的 ORM 查询
|
|
tokens = self.session.query(Token).filter(
|
|
Token.symbol.like(f"{key}%") | Token.base_asset.like(f"{key}%")
|
|
).limit(limit).all()
|
|
|
|
return [{
|
|
'symbol': token.symbol,
|
|
'base_asset': token.base_asset,
|
|
'quote_asset': token.quote_asset
|
|
} for token in tokens]
|
|
|
|
except Exception as e:
|
|
logger.error(f"搜索Token信息失败: {e}")
|
|
return [] |