Python获取Binance流数据存入数据库后价格精度丢失求助

使用Python的unicorn_binance_websocket_api获取Binance的BTCUSDT等币种1分钟K线流数据，打印时价格显示完整格式（如"26738.94000000"），但存入MariaDB和SQLite数据库后，开盘价、收盘价、最高价、最低价等字段被截断为两位小数。手动插入完整精度的数据可正常存储，排查代码后仍未找到问题根源，寻求技术指导。

数据示例 #

{"stream_type":"btcusdt@kline_1m","event_type":"kline","event_time":1686983029769,"symbol":"BTCUSDT","kline":{"kline_start_time":1686982980000,"kline_close_time":1686983039999,"symbol":"BTCUSDT","interval":"1m","first_trade_id":false,"last_trade_id":false,"open_price":"26738.94000000","close_price":"26777.99000000","high_price":"26778.00000000","low_price":"26738.94000000","base_volume":"52.18289000","number_of_trades":1613,"is_closed":false,"quote":"1396485.64284410","taker_by_base_asset_volume":"37.43018000","taker_by_quote_asset_volume":"1001618.64984080","ignore":"0"},"unicorn_fied":["binance.com","0.12.2"]}

代码片段 #

# import
from unicorn_binance_websocket_api.manager import BinanceWebSocketApiManager
import time 
import pandas as pd
import sqlalchemy
from sqlalchemy import create_engine

# create engine to connect to DB
engine = sqlalchemy.create_engine("mariadb+mariadbconnector://test_user:StrongPassword@127.0.0.1:3306/test_db")

# define coins to collect data from and start stream
symbols = ['BTC','COMBO']
symbols = [symbol+'usdt' for symbol in symbols]

ubwa = BinanceWebSocketApiManager(exchange="binance.com", output_default="UnicornFy")
ubwa.create_stream(['kline_1m'], symbols, output="UnicornFy")

# sort data from stream into dataframes and insert into db
def SQLimport(data):
    time = data['event_time']
    coin = data['symbol']
    open_price = data['kline']['open_price']
    close_price = data['kline']['close_price']
    low_price = data['kline']['low_price']
    high_price = data['kline']['high_price']
    volume = data['kline']['taker_by_base_asset_volume']
    frame = pd.DataFrame([[time,open_price,close_price,low_price,high_price,volume]], columns = ['time','open_price','close_price','low_price','high_price','volume'])
    frame.time = pd.to_datetime(frame.time, unit='ms')
    frame.open_price = frame.open_price.astype(float)
    frame.close_price = frame.close_price.astype(float)
    frame.low_price = frame.low_price.astype(float)
    frame.high_price = frame.high_price.astype(float)
    frame.volume = frame.volume.astype(float)
    frame['amplitude'] = (frame.high_price - frame.low_price) / (frame.high_price + frame.low_price) /2
    frame.to_sql(coin, engine, index=False, if_exists='append')

while True:
    data = ubwa.pop_stream_data_from_stream_buffer()

    if data:
        if len(data) > 3:
            SQLimport(data)
            print(data)

问题核心是Pandas自动创建数据库表时，默认将float类型映射为仅保留两位小数的DECIMAL类型，导致高精度数值被截断。手动插入正常是因为数据库字段本身支持高精度，只是Pandas自动建表的类型推断不符合需求。

1. 提前创建高精度数据库表 #

在MariaDB/SQLite中预先创建表，指定字段为支持Binance精度的类型（比如DECIMAL(18,8)，匹配Binance的8位小数价格）：

-- MariaDB 示例
CREATE TABLE BTCUSDT (
    time DATETIME,
    open_price DECIMAL(18,8),
    close_price DECIMAL(18,8),
    low_price DECIMAL(18,8),
    high_price DECIMAL(18,8),
    volume DECIMAL(18,8),
    amplitude DECIMAL(18,10)
);

-- SQLite 示例
CREATE TABLE BTCUSDT (
    time DATETIME,
    open_price NUMERIC(18,8),
    close_price NUMERIC(18,8),
    low_price NUMERIC(18,8),
    high_price NUMERIC(18,8),
    volume NUMERIC(18,8),
    amplitude NUMERIC(18,10)
);

2. 在to_sql中指定字段类型映射 #

如果需要Pandas自动处理表创建，可通过dtype参数手动指定SQLAlchemy字段类型，避免自动推断错误：
首先导入SQLAlchemy类型：

from sqlalchemy.types import DECIMAL, DateTime

修改SQLimport函数中的to_sql调用：

def SQLimport(data):
    # ... 原有代码 ...
    frame['amplitude'] = (frame.high_price - frame.low_price) / (frame.high_price + frame.low_price) /2
    
    # 指定字段类型映射
    dtype = {
        'time': DateTime(),
        'open_price': DECIMAL(18,8),
        'close_price': DECIMAL(18,8),
        'low_price': DECIMAL(18,8),
        'high_price': DECIMAL(18,8),
        'volume': DECIMAL(18,8),
        'amplitude': DECIMAL(18,10)
    }
    frame.to_sql(coin, engine, index=False, if_exists='append', dtype=dtype)

3. 用Decimal类型替代float避免精度损失 #

Binance返回的价格是字符串格式的高精度数值，转成float可能存在隐性精度损失，建议直接用decimal.Decimal处理：

from decimal import Decimal

def SQLimport(data):
    # ... 原有代码 ...
    frame.time = pd.to_datetime(frame.time, unit='ms')
    # 替换astype(float)为Decimal转换
    frame.open_price = frame.open_price.apply(Decimal)
    frame.close_price = frame.close_price.apply(Decimal)
    frame.low_price = frame.low_price.apply(Decimal)
    frame.high_price = frame.high_price.apply(Decimal)
    frame.volume = frame.volume.apply(Decimal)
    frame['amplitude'] = (frame.high_price - frame.low_price) / (frame.high_price + frame.low_price) /2
    
    # 配合dtype参数存入数据库
    dtype = {
        'time': DateTime(),
        'open_price': DECIMAL(18,8),
        'close_price': DECIMAL(18,8),
        'low_price': DECIMAL(18,8),
        'high_price': DECIMAL(18,8),
        'volume': DECIMAL(18,8),
        'amplitude': DECIMAL(18,10)
    }
    frame.to_sql(coin, engine, index=False, if_exists='append', dtype=dtype)

内容的提问来源于stack exchange，提问作者John Spencer