一文详解Python如何同时迭代多个序列

# 基本zip使用
names = ['Alice', 'Bob', 'Charlie']
ages = [25, 30, 35]
salaries = [50000, 60000, 70000]

print("基本zip迭代:")
for name, age, salary in zip(names, ages, salaries):
    print(f"{name}: {age}岁, 薪资: ${salary}")

# 输出:
# Alice: 25岁, 薪资: $50000
# Bob: 30岁, 薪资: $60000
# Charlie: 35岁, 薪资: $70000

from itertools import zip_longest

# 不等长序列
students = ['Alice', 'Bob', 'Charlie', 'David']
scores = [90, 85, 92]

print("\nzip_longest迭代:")
for student, score in zip_longest(students, scores, fillvalue='缺考'):
    print(f"{student}: {score}")

# 输出:
# Alice: 90
# Bob: 85
# Charlie: 92
# David: 缺考

def parallel_process(iterables, func):
    """多序列并行处理"""
    for args in zip(*iterables):
        yield func(*args)

# 使用示例
a = [1, 2, 3]
b = [4, 5, 6]
c = [7, 8, 9]

result = parallel_process([a, b, c], lambda x, y, z: x*y + z)
print("并行处理结果:", list(result))  # [11, 18, 27]

def multi_filter(iterables, condition):
    """多序列条件过滤"""
    for args in zip(*iterables):
        if condition(*args):
            yield args

# 使用示例
temperatures = [22.1, 23.5, 21.8, 25.3, 20.6]
humidities = [45, 60, 55, 70, 50]

# 筛选高温高湿数据
filtered = multi_filter([temperatures, humidities], 
                        lambda t, h: t > 23 and h > 55)
print("高温高湿数据:")
for t, h in filtered:
    print(f"温度: {t}℃, 湿度: {h}%")  # (23.5, 60), (25.3, 70)

def align_time_series(series1, series2, time_key='timestamp'):
    """时间序列对齐"""
    # 创建时间索引映射
    index_map = {}
    for item in series1:
        index_map[item[time_key]] = item
    
    # 对齐序列
    for item in series2:
        ts = item[time_key]
        if ts in index_map:
            yield index_map[ts], item

# 使用示例
temp_data = [
    {'timestamp': '2023-01-01 10:00', 'temp': 22.1},
    {'timestamp': '2023-01-01 11:00', 'temp': 23.5},
    {'timestamp': '2023-01-01 12:00', 'temp': 25.3}
]

hum_data = [
    {'timestamp': '2023-01-01 10:00', 'hum': 45},
    {'timestamp': '2023-01-01 11:00', 'hum': 60},
    {'timestamp': '2023-01-01 12:00', 'hum': 55}
]

print("时间序列对齐:")
for temp, hum in align_time_series(temp_data, hum_data):
    print(f"时间: {temp['timestamp']}, 温度: {temp['temp']}, 湿度: {hum['hum']}")

def resample_align(series1, series2, freq='H'):
    """重采样对齐时间序列"""
    import pandas as pd
    
    # 创建DataFrame
    df1 = pd.DataFrame(series1).set_index('timestamp')
    df2 = pd.DataFrame(series2).set_index('timestamp')
    
    # 重采样
    df1_resampled = df1.resample(freq).mean()
    df2_resampled = df2.resample(freq).mean()
    
    # 对齐
    aligned = pd.concat([df1_resampled, df2_resampled], axis=1).dropna()
    return aligned.iterrows()

# 使用示例
# 假设有更多数据点
aligned = resample_align(temp_data, hum_data)
print("\n重采样对齐:")
for ts, row in aligned:
    print(f"时间: {ts}, 温度: {row['temp']}, 湿度: {row['hum']}")

def feature_engineering(features, labels):
    """多序列特征工程"""
    for feature_vec, label in zip(features, labels):
        # 创建新特征
        new_feature = sum(feature_vec) / len(feature_vec)  # 平均值
        yield feature_vec + [new_feature], label

# 使用示例
features = [
    [1.2, 3.4, 5.6],
    [2.3, 4.5, 6.7],
    [3.4, 5.6, 7.8]
]
labels = [0, 1, 0]

print("特征工程结果:")
for new_features, label in feature_engineering(features, labels):
    print(f"特征: {new_features}, 标签: {label}")

def merge_datasets(datasets, key='id'):
    """多数据集合并"""
    # 创建主索引
    master_index = {}
    for dataset in datasets:
        for item in dataset:
            item_key = item[key]
            if item_key not in master_index:
                master_index[item_key] = {}
            master_index[item_key].update(item)
    
    # 生成合并数据
    for key, data in master_index.items():
        yield data

# 使用示例
users = [{'id': 1, 'name': 'Alice'}, {'id': 2, 'name': 'Bob'}]
orders = [{'id': 1, 'order': 'A123'}, {'id': 2, 'order': 'B456'}]
payments = [{'id': 1, 'amount': 100}, {'id': 2, 'amount': 200}]

print("数据合并结果:")
for merged in merge_datasets([users, orders, payments]):
    print(merged)

class SensorFusion:
    """多传感器数据融合系统"""
    def __init__(self, sensors):
        self.sensors = sensors
    
    def read(self):
        """读取并融合传感器数据"""
        # 同时读取所有传感器
        readings = [sensor.read() for sensor in self.sensors]
        
        # 简单融合：平均值
        avg_value = sum(readings) / len(readings)
        return avg_value, readings
    
    def continuous_fusion(self):
        """连续数据融合"""
        while True:
            yield self.read()

# 传感器模拟
class MockSensor:
    def __init__(self, name):
        self.name = name
        self.value = 0
    
    def read(self):
        self.value += 1
        return self.value

# 使用示例
sensor1 = MockSensor('温度')
sensor2 = MockSensor('湿度')
fusion = SensorFusion([sensor1, sensor2])

print("传感器融合:")
for _ in range(3):
    avg, readings = fusion.read()
    print(f"平均值: {avg}, 原始值: {readings}")

def multi_stream_processor(streams, process_func):
    """多数据流处理器"""
    iterators = [iter(stream) for stream in streams]
    
    while True:
        try:
            items = [next(it) for it in iterators]
            yield process_func(items)
        except StopIteration:
            break

# 使用示例
stream1 = (i for i in range(5))  # 生成器流
stream2 = (i*2 for i in range(5))

processor = multi_stream_processor(
    [stream1, stream2],
    lambda items: sum(items)
)

print("多流处理结果:", list(processor))  # [0, 3, 6, 9, 12]

def multimodal_training(image_stream, text_stream, label_stream):
    """多模态训练数据生成"""
    for image, text, label in zip(image_stream, text_stream, label_stream):
        # 多模态特征融合
        fused_features = fuse_features(image, text)
        yield fused_features, label

def fuse_features(image, text):
    """特征融合（示例）"""
    # 实际应用中会使用深度学习模型
    return f"图像特征:{image[:10]}... + 文本特征:{text[:10]}..."

# 使用示例
images = ['image_data1', 'image_data2', 'image_data3']
texts = ['文本描述1', '文本描述2', '文本描述3']
labels = [0, 1, 0]

print("多模态训练数据:")
for features, label in multimodal_training(images, texts, labels):
    print(f"特征: {features}, 标签: {label}")

def cross_validation_generator(X, y, folds=5):
    """交叉验证生成器"""
    from sklearn.model_selection import KFold
    kf = KFold(n_splits=folds)
    
    for train_index, test_index in kf.split(X):
        X_train, X_test = X[train_index], X[test_index]
        y_train, y_test = y[train_index], y[test_index]
        yield X_train, X_test, y_train, y_test

# 使用示例
import numpy as np
X = np.array([[1, 2], [3, 4], [5, 6], [7, 8], [9, 10]])
y = np.array([0, 1, 0, 1, 0])

print("交叉验证折叠:")
for fold, (X_train, X_test, y_train, y_test) in enumerate(cross_validation_generator(X, y)):
    print(f"折叠 {fold+1}:")
    print(f"训练集: {X_train}, 标签: {y_train}")
    print(f"测试集: {X_test}, 标签: {y_test}")

import threading
from queue import Queue

def concurrent_data_collection(sources):
    """多线程数据采集"""
    results = Queue()
    threads = []
    
    def worker(source, output):
        """数据采集工作线程"""
        data = source.collect()
        output.put(data)
    
    for source in sources:
        t = threading.Thread(target=worker, args=(source, results))
        t.start()
        threads.append(t)
    
    # 等待所有线程完成
    for t in threads:
        t.join()
    
    # 返回结果
    return [results.get() for _ in sources]

# 数据源模拟
class DataSource:
    def __init__(self, name):
        self.name = name
    
    def collect(self):
        import time
        time.sleep(1)  # 模拟采集延迟
        return f"{self.name}数据"

# 使用示例
sources = [DataSource(f"源{i+1}") for i in range(3)]
data = concurrent_data_collection(sources)
print("并发采集数据:", data)

import asyncio

async def async_multi_iter(iterables, process_func):
    """异步多序列迭代"""
    iterators = [iter(iterable) for iterable in iterables]
    
    while True:
        try:
            # 异步获取下一组数据
            tasks = [asyncio.to_thread(next, it) for it in iterators]
            results = await asyncio.gather(*tasks, return_exceptions=True)
            
            # 检查是否有迭代结束
            if any(isinstance(res, StopIteration) for res in results):
                break
            
            # 处理数据
            yield process_func(results)
        except StopIteration:
            break

# 使用示例
async def main():
    streams = [
        [1, 2, 3, 4],
        ['a', 'b', 'c', 'd'],
        [0.1, 0.2, 0.3, 0.4]
    ]
    
    async for result in async_multi_iter(streams, lambda items: tuple(items)):
        print("异步结果:", result)

asyncio.run(main())

def memory_efficient_zip(iterables):
    """内存高效多序列迭代"""
    iterators = [iter(it) for it in iterables]
    while True:
        try:
            yield tuple(next(it) for it in iterators)
        except StopIteration:
            break

# 使用示例
large_stream1 = (i for i in range(1000000))
large_stream2 = (i*2 for i in range(1000000))

print("内存高效迭代:")
for i, (a, b) in enumerate(memory_efficient_zip([large_stream1, large_stream2])):
    if i >= 3:  # 只显示前3个
        break
    print(a, b)

def chunked_multi_process(iterables, process_func, chunk_size=1000):
    """分块多序列处理"""
    iterators = [iter(it) for it in iterables]
    while True:
        chunk = []
        try:
            for _ in range(chunk_size):
                items = tuple(next(it) for it in iterators)
                chunk.append(items)
        except StopIteration:
            if not chunk:
                break
        
        # 处理整个块
        processed = process_func(chunk)
        yield from processed

# 使用示例
def process_chunk(chunk):
    """处理数据块（示例）"""
    return [sum(items) for items in chunk]

stream1 = range(10000)
stream2 = range(10000, 20000)

print("分块处理结果:")
results = chunked_multi_process([stream1, stream2], process_chunk, chunk_size=3)
for i, res in enumerate(results):
    if i >= 5:  # 只显示前5个
        break
    print(res)  # 0+10000=10000, 1+10001=10002, 2+10002=10004, ...

# 等长序列
for a, b in zip(seq1, seq2):
    process(a, b)

# 不等长序列
for a, b in zip_longest(seq1, seq2, fillvalue=None):
    process(a, b)

# 时间序列
for ts, (a, b) in align_time_series(seq1, seq2):
    process(ts, a, b)

# 使用生成器避免内存问题
stream1 = (x for x in large_dataset1)
stream2 = (x for x in large_dataset2)
for a, b in zip(stream1, stream2):
    process(a, b)

def safe_multi_iter(iterables):
    """安全的多序列迭代"""
    iterators = [iter(it) for it in iterables]
    while True:
        try:
            items = []
            for it in iterators:
                items.append(next(it))
            yield tuple(items)
        except StopIteration:
            break
        except Exception as e:
            print(f"迭代错误: {e}")
            # 处理错误或跳过
            continue

# 避免不必要的转换
# 错误做法
for a, b in zip(list1, list2):
    ...

# 正确做法
for a, b in zip(iter1, iter2):
    ...

def align_by_key(iterables, key_func):
    """按键函数对齐序列"""
    # 创建索引映射
    index_map = {}
    for it in iterables:
        for item in it:
            key = key_func(item)
            if key not in index_map:
                index_map[key] = [None] * len(iterables)
            index_map[key][iterables.index(it)] = item

    # 生成对齐数据
    for key, items in index_map.items():
        if all(item is not None for item in items):
            yield items

def multi_sequence_processor(sequences, process_func):
    """
    多序列处理器

    参数:
        sequences: 序列列表
        process_func: 处理函数，接受每个序列的一个元素

    返回:
        处理结果的生成器

    注意:
        所有序列长度应相同，否则截断到最短序列
    """
    for items in zip(*sequences):
        yield process_func(*items)