掌握数据库领域，运用时序数据库工具

    import time
    from datetime import datetime, timedelta
    
    def time_partition(timestamp, interval='day'):
    """将时间戳按指定间隔分区"""
    dt = datetime.fromtimestamp(timestamp)
    if interval == 'hour':
        return dt.replace(minute=0, second=0, microsecond=0)
    elif interval == 'day':
        return dt.replace(hour=0, minute=0, second=0, microsecond=0)
    elif interval == 'month':
        return dt.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
    else:
        return dt
    
    # 示例使用
    current_time = time.time()
    print(f"当前时间分区(小时): {time_partition(current_time, 'hour')}")
    print(f"当前时间分区(天): {time_partition(current_time, 'day')}")
    print(f"当前时间分区(月): {time_partition(current_time, 'month')}")
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/0SCBH4DpKY9latrQmjvqPkEOs1MZ.png)

    import numpy as np
    
    def compress_timeseries(timestamps, values, tolerance=0.01):
    """使用死区压缩算法简化时间序列"""
    compressed_ts = [timestamps[0]]
    compressed_val = [values[0]]
    last_val = values[0]
    
    for ts, val in zip(timestamps[1:], values[1:]):
        if abs(val - last_val) > tolerance:
            compressed_ts.append(ts)
            compressed_val.append(val)
            last_val = val
    
    return np.array(compressed_ts), np.array(compressed_val)
    
    # 示例使用
    timestamps = np.arange(0, 100, 1)
    values = np.sin(timestamps * 0.1)
    compressed_ts, compressed_val = compress_timeseries(timestamps, values, 0.05)
    
    print(f"原始数据点: {len(timestamps)}")
    print(f"压缩后数据点: {len(compressed_ts)}")
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/pIiaJTVSuyrql2t1x6AnweKdmMZ3.png)

    class TimeIndex:
    """简单的时间索引实现"""
    def __init__(self):
        self.min_timestamp = float('inf')
        self.max_timestamp = -float('inf')
        self.time_blocks = {}
    
    def add_record(self, timestamp, record_id):
        """添加记录到时间索引"""
        self.min_timestamp = min(self.min_timestamp, timestamp)
        self.max_timestamp = max(self.max_timestamp, timestamp)
    
        # 按小时分区
        block_key = int(timestamp // 3600)
        if block_key not in self.time_blocks:
            self.time_blocks[block_key] = []
        self.time_blocks[block_key].append((timestamp, record_id))
    
    def query_range(self, start, end):
        """查询时间范围内的记录"""
        result = []
        start_block = int(start // 3600)
        end_block = int(end // 3600)
    
        for block in range(start_block, end_block + 1):
            if block in self.time_blocks:
                for ts, record_id in self.time_blocks[block]:
                    if start <= ts <= end:
                        result.append(record_id)
        return result
    
    # 示例使用
    index = TimeIndex()
    for i in range(1000):
    index.add_record(1609459200 + i * 60, f"record_{i}")  # 假设从2021-01-01开始每分钟一条数据
    
    print("索引时间范围:", index.min_timestamp, "to", index.max_timestamp)
    print("查询结果示例:", index.query_range(1609459200, 1609459200 + 3600)[:5])
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/yVvdz4woQXRtbNUhECMFg6mIpHr3.png)

    # 使用Docker快速启动InfluxDB
    docker run -d -p 8086:8086 \
      -v influxdb:/var/lib/influxdb \
      influxdb:latest
    
    # 创建管理员用户
    docker exec -it influxdb influx
    > CREATE USER admin WITH PASSWORD 'password' WITH ALL PRIVILEGES
    > CREATE DATABASE metrics
    > USE metrics
    
    
    bash
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/nz8xoYV36jLhJckBmO9G0ftIb2MX.png)

    # 使用Docker启动TimescaleDB
    docker run -d -p 5432:5432 \
      -v timescaledb:/var/lib/postgresql/data \
      -e POSTGRES_PASSWORD=password \
      timescale/timescaledb:latest-pg12
    
    # 连接到数据库并创建扩展
    psql -U postgres -h localhost
    CREATE DATABASE metrics;
    \c metrics
    CREATE EXTENSION IF NOT EXISTS timescaledb CASCADE;
    
    
    bash
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/7G8CnLv1qRSmtihke0aHY9xpTBsf.png)

    from influxdb_client import InfluxDBClient, Point, WriteOptions
    from influxdb_client.client.write_api import SYNCHRONOUS
    
    # 配置客户端
    client = InfluxDBClient(url="http://localhost:8086", token="admin:password", org="-")
    write_api = client.write_api(write_options=SYNCHRONOUS)
    query_api = client.query_api()
    
    # 写入数据
    point = Point("temperature").tag("location", "server_room").field("value", 25.3).time(time.time_ns())
    write_api.write(bucket="metrics", record=point)
    
    # 查询数据
    query = 'from(bucket:"metrics") |> range(start:-1h) |> filter(fn: (r) => r._measurement == "temperature")'
    result = query_api.query(query)
    for table in result:
    for record in table.records:
        print(f"{record.get_time()}: {record.get_value()}")
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/2GeMTqJlOapSZfHtb86C1P0N9g4Q.png)

    import psycopg2
    from datetime import datetime, timedelta
    import random
    
    # 连接数据库
    conn = psycopg2.connect(
    host="localhost",
    database="metrics",
    user="postgres",
    password="password"
    )
    cursor = conn.cursor()
    
    # 创建时序表
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS cpu_metrics (
    time TIMESTAMPTZ NOT NULL,
    device_id TEXT NOT NULL,
    cpu_usage DOUBLE PRECISION NOT NULL
    );
    """)
    
    # 转换为超表
    cursor.execute("SELECT create_hypertable('cpu_metrics', 'time');")
    conn.commit()
    
    # 插入示例数据
    now = datetime.now()
    for i in range(100):
    time = now - timedelta(minutes=i)
    device = f"device_{random.randint(1, 5)}"
    usage = random.uniform(0.1, 0.9)
    cursor.execute(
        "INSERT INTO cpu_metrics (time, device_id, cpu_usage) VALUES (%s, %s, %s)",
        (time, device, usage)
    )
    conn.commit()
    
    # 查询最近一小时的数据
    cursor.execute("""
    SELECT time_bucket('5 minutes', time) AS interval,
       avg(cpu_usage) AS avg_usage,
       device_id
    FROM cpu_metrics
    WHERE time > now() - INTERVAL '1 hour'
    GROUP BY interval, device_id
    ORDER BY interval DESC;
    """)
    
    for row in cursor.fetchall():
    print(f"时间区间: {row[0]}, 设备: {row[2]}, 平均使用率: {row[1]:.2%}")
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/Bjlhqcuwysp7ISinMVa85x1PE9fC.png)

 * 自动扩展和弹性部署能力
 * 多区域复制和全球可用性

 * 内置异常检测算法
 * 自动化预测和模式识别

 * 轻量级时序数据库边缘版本
 * 边缘-云端数据同步

 * 时序数据与图数据、文档数据结合
 * 统一查询接口

 * 海量历史数据的高效存储和检索
 * 冷热数据分层存储策略

 * 平衡简单查询和复杂分析需求
 * 跨时间线关联查询优化

 * 处理不完整或不规则的时间序列
 * 自动数据修复和插值