cryptoservice.utils

`cryptoservice.utils`

Classes

`CacheManager(ttl_seconds: int = 60)`

缓存管理器.

Source code in src/cryptoservice/utils/cache_manager.py

def __init__(self, ttl_seconds: int = 60):
    self._cache: Dict[str, Tuple[Any, datetime]] = {}
    self._ttl = ttl_seconds
    self._lock = threading.Lock()

Functions

`get(key: str) -> Optional[Any]`

获取缓存数据.

Source code in src/cryptoservice/utils/cache_manager.py

def get(self, key: str) -> Optional[Any]:
    """获取缓存数据."""
    with self._lock:
        if key in self._cache:
            data, timestamp = self._cache[key]
            if datetime.now() - timestamp < timedelta(seconds=self._ttl):
                return data
            del self._cache[key]
    return None

`set(key: str, value: Any) -> None`

设置缓存数据.

Source code in src/cryptoservice/utils/cache_manager.py

def set(self, key: str, value: Any) -> None:
    """设置缓存数据."""
    with self._lock:
        self._cache[key] = (value, datetime.now())

`clear() -> None`

清除所有缓存.

Source code in src/cryptoservice/utils/cache_manager.py

def clear(self) -> None:
    """清除所有缓存."""
    with self._lock:
        self._cache.clear()

`DataConverter`

数据转换工具类.

Functions

`to_decimal(value: Union[str, float, int]) -> Decimal` `staticmethod`

转换为Decimal类型.

Source code in src/cryptoservice/utils/data_converter.py

@staticmethod
def to_decimal(value: Union[str, float, int]) -> Decimal:
    """转换为Decimal类型."""
    return Decimal(str(value))

`format_timestamp(timestamp: Union[int, float]) -> datetime` `staticmethod`

转换时间戳为datetime对象.

Source code in src/cryptoservice/utils/data_converter.py

@staticmethod
def format_timestamp(timestamp: Union[int, float]) -> datetime:
    """转换时间戳为datetime对象."""
    if isinstance(timestamp, (int, float)):
        return datetime.fromtimestamp(timestamp / 1000)
    return datetime.now()

`format_market_data(data: Dict[str, Any]) -> Dict[str, Any]` `staticmethod`

格式化市场数据.

Source code in src/cryptoservice/utils/data_converter.py

@staticmethod
def format_market_data(data: Dict[str, Any]) -> Dict[str, Any]:
    """格式化市场数据."""
    return {
        "price": float(data.get("price", 0)),
        "volume": float(data.get("volume", 0)),
        "change": float(data.get("priceChangePercent", 0)),
        "high": float(data.get("highPrice", 0)),
        "low": float(data.get("lowPrice", 0)),
        "timestamp": datetime.now().isoformat(),
    }

Functions

`print_table(data: list[Any], title: str | None = None, headers: list[str] | None = None) -> None`

打印表格数据.

PARAMETER	DESCRIPTION
`data`	表格数据 TYPE: `list[Any]`
`title`	表格标题 TYPE: `str \| None` DEFAULT: `None`
`headers`	列标题列表，如果为None则自动生成 TYPE: `list[str] \| None` DEFAULT: `None`

RAISES	DESCRIPTION
`ValueError`	当数据为空或格式不正确时

Source code in src/cryptoservice/utils/logger.py

def print_table(
    data: list[Any],
    title: str | None = None,
    headers: list[str] | None = None,
) -> None:
    """打印表格数据.

    Args:
        data: 表格数据
        title: 表格标题
        headers: 列标题列表，如果为None则自动生成

    Raises:
        ValueError: 当数据为空或格式不正确时
    """
    try:
        # 检查数据是否为空
        if not data:
            raise ValueError("Empty data provided")

        # 检查数据是否为列表
        if not isinstance(data, list):
            raise ValueError(f"Expected list, got {type(data).__name__}")

        table = Table(show_header=True, header_style="bold magenta")

        # 如果数据是字典列表
        if isinstance(data[0], dict):
            # 验证所有行是否都是字典
            if not all(isinstance(row, dict) for row in data):
                raise ValueError("Inconsistent row types in dictionary data")

            headers = headers or list(data[0].keys())
            for header in headers:
                table.add_column(header, style="cyan")
            for row in data:
                # 检查是否所有必需的键都存在
                missing_keys = set(headers) - set(row.keys())
                if missing_keys:
                    print_error(f"Missing keys in row: {missing_keys}")
                table.add_row(*[str(row.get(h, "N/A")) for h in headers])

        # 如果数据是普通列表
        else:
            # 验证所有行的长度是否一致
            row_lengths = {len(row) if isinstance(row, (list, tuple)) else 1 for row in data}
            if len(row_lengths) > 1:
                raise ValueError(f"Inconsistent row lengths: {row_lengths}")

            row_length = row_lengths.pop()
            headers = headers or [f"Column {i + 1}" for i in range(row_length)]

            # 验证headers长度是否匹配数据
            if len(headers) != row_length:
                raise ValueError(f"Headers length ({len(headers)}) doesn't match data width ({row_length})")

            for header in headers:
                table.add_column(header, style="cyan")
            for row in data:
                if not isinstance(row, (list, tuple)):
                    row = [row]  # 单个值转换为列表
                table.add_row(*[str(x) for x in row])

        if title:
            console.print(f"\n[bold]{title}[/bold]")
        console.print(table)

    except Exception as e:
        print_error(f"Failed to print table: {str(e)}")
        raise

cryptoservice.utils