علامت lag در هم‌بستگی سیگنال با پایتون؛ مرجع دستی و آزمون مستقل

در بازبینی برنامهٔ پردازش سیگنال، خروجی هم‌بستگی ممکن است درست به نظر برسد ولی علامت جابه‌جایی برعکس گزارش شود. این خطا معمولاً از نامشخص بودن ترتیب دو ورودی یا یکی‌گرفتن اندیس آرایه با lag می‌آید. در این درس، دو دنبالهٔ ولتاژ ساختگی را با قرارداد روشن، پاسخ دستی و مرجع مستقل بررسی می‌کنیم. اندازه‌گیری تأخیر واقعی، هم‌زمان‌سازی دستگاه و اثبات رابطهٔ علّی از این تمرین به دست نمی‌آید.

۱. پیش از کدنویسی، دو ورودی و علامت را تعریف کنیم

قرارداد ما برای دادهٔ حقیقی r_xy[k]=Σx[n]y[n−k] است. ابتدا x و سپس y داده می‌شود؛ جابه‌جایی این ترتیب معنای علامت را تغییر می‌دهد. بخش Notes در correlate از SciPy همین فرمول را با مزدوج مختلط ورودی دوم توضیح می‌دهد. این برنامه فقط دادهٔ حقیقی می‌پذیرد و حالت مختلط را پیاده نمی‌کند. SciPy نیز در اجرای ثبت‌شده استفاده نشده است.

اگر طول x برابر N و طول y برابر M باشد، lag از منفی M−1 تا N−1، شامل دو انتهاست؛ تعداد خروجی N+M−1 است. خارج اندیس‌های معتبر هر دنباله، مقدار صفر فرض می‌شود. این امتداد صفر یک قرارداد ریاضی برای دادهٔ محدود است؛ اندازه‌گیری تازه یا رفتار تکرارشوندهٔ سیگنال نیست. نمونهٔ انتهای آرایه به ابتدای آن پیچیده نمی‌شود، پس محاسبه خطی است و دوری نیست.

راهنمای NumPy correlate دربارهٔ وجود تعریف‌های متفاوت با علامت مخالف هشدار می‌دهد. بنابراین نتیجه‌ای مانند «lag مثبت یعنی ورودی دوم دیرتر است» را بدون فرمول و ترتیب ورودی‌ها حفظ نمی‌کنیم. حتی وقتی اسم تابع‌ها یکسان است، پیش‌فرض mode و قرارداد گزارش باید بررسی شود. در این تمرین فقط full تعریف‌شدهٔ خودمان وجود دارد.

۲. هم‌پوشانی محدود و واحد پاسخ خام

در هر lag، فقط زوج‌هایی ضرب می‌شوند که هر دو اندیس داخل آرایه باشند؛ بقیه سهم صفر دارند. شمار این زوج‌ها در لبه‌ها کمتر از بخش میانی است. نمونهٔ صفرِ موجود در آرایه همچنان عضو معتبر است، هرچند حاصل‌ضربش صفر شود. این شمارش را جدا ثبت می‌کنیم تا «تعداد هم‌پوشانی» با تعداد مقدارهای غیرصفر اشتباه نشود.

هر دو ورودی واحد ولت دارند؛ حاصل‌ضرب و جمع خام واحد V² دارد. پاسخ می‌تواند از یک بزرگ‌تر یا منفی باشد و ضریب بی‌بعدِ Pearson نیست. میانگین حذف نشده و تقسیم بر انرژی یا تعداد هم‌پوشانی انجام نمی‌شود. پس بهره، مؤلفهٔ ثابت و طول هم‌پوشانی می‌توانند مقدارها و تفسیر قله را تحت تأثیر بگذارند. نرمال‌سازی‌های دیگر، تعریف و آزمون جدا می‌خواهند.

برای گزارش قله، بزرگ‌ترین مقدار علامت‌دار را انتخاب می‌کنیم، نه بزرگ‌ترین قدرمطلق. در برابری دقیق floatها، کوچک‌ترین lag انتخاب و نبود یکتایی علامت زده می‌شود. این سیاست صرفاً نتیجه را قطعی می‌کند؛ شاهد کیفیت تطبیق نیست. وقتی همهٔ خروجی‌ها صفرند، lag انتخاب‌شده از همین سیاست می‌آید و تأخیر معنی‌داری ثابت نمی‌کند. نزدیک بودن دو قله نیز در این نسخه تحلیل آماری نمی‌شود.

۳. کد کامل و تفاوت اندیس قله با lag

ورودی باید list یا tuple غیرخالی از int یا float دقیق و متناهی باشد. bool، مقدار مختلط، رشته، NaN و بی‌نهایت رد می‌شوند؛ واحد هر دو ورودی دقیقاً V است. پس از تبدیل به float، متناهی بودن حاصل‌ضرب‌ها و جمع نیز بررسی می‌شود. کد با Python 3.14.5 و کتابخانهٔ استاندارد اجرا شده؛ NumPy یا SciPy اجرا نشده‌اند و سرعت کتابخانه‌ها را مقایسه نمی‌کنیم.

"""Original real-valued full linear cross-correlation; no measured signals."""
from fractions import Fraction
import itertools
import json
import math
import platform


def _finite_number(value, field):
    if type(value) not in (int, float):
        raise TypeError(field + ' must be exact int or float')
    try:
        value = float(value)
    except OverflowError as error:
        raise ValueError(field + ' cannot be represented as finite float') from error
    if not math.isfinite(value):
        raise ValueError(field + ' must be finite')
    return value


def _samples(values, field):
    if type(values) not in (list, tuple):
        raise TypeError(field + ' must be finite list or tuple')
    if not values:
        raise ValueError(field + ' must be nonempty')
    return [_finite_number(value, field) for value in values]


def cross_correlate(x, y, unit='V'):
    """r_xy[k] = sum_n x[n] * y[n-k], zero outside each finite support."""
    if type(unit) is not str or unit != 'V':
        raise ValueError('both signal units must be V')
    first, second = _samples(x, 'x'), _samples(y, 'y')
    n_count, m_count = len(first), len(second)
    lags = list(range(-(m_count - 1), n_count))
    scores, overlaps = [], []
    for lag in lags:
        start, stop = max(0, lag), min(n_count, m_count + lag)
        products = [first[n] * second[n - lag] for n in range(start, stop)]
        if not all(math.isfinite(product) for product in products):
            raise ArithmeticError('nonfinite cross-correlation product')
        try:
            score = math.fsum(products)
        except OverflowError as error:
            raise ArithmeticError('cross-correlation summation overflow') from error
        if not math.isfinite(score):
            raise ArithmeticError('nonfinite cross-correlation score')
        scores.append(score)
        overlaps.append(stop - start)
    maximum = max(scores)  # signed maximum, not maximum absolute value
    tied_indices = [index for index, score in enumerate(scores) if score == maximum]
    chosen = tied_indices[0]  # ascending lags: exact ties choose the smallest lag
    return {'lags': lags, 'correlation_V2': scores, 'overlap_samples': overlaps,
            'peak': {'index': chosen, 'lag_samples': lags[chosen], 'correlation_V2': maximum,
                     'unique': len(tied_indices) == 1, 'exact_ties': len(tied_indices)}}


def lag_seconds(lag_samples, sample_rate, unit='Hz'):
    if type(lag_samples) is not int:
        raise TypeError('lag must be exact int samples')
    if type(unit) is not str or unit != 'Hz':
        raise ValueError('sample rate unit must be Hz')
    sample_rate = _finite_number(sample_rate, 'sample_rate')
    if not sample_rate > 0:
        raise ValueError('sample_rate must be positive')
    try:
        seconds = lag_samples / sample_rate
    except OverflowError as error:
        raise ArithmeticError('lag seconds overflow') from error
    if not math.isfinite(seconds):
        raise ArithmeticError('nonfinite lag seconds')
    return seconds


def reference_reversed_convolution(x, y):
    """Independent exact scatter convolution of x with reversed real y."""
    result = [Fraction(0) for _ in range(len(x) + len(y) - 1)]
    for i, a in enumerate(x):
        for j, b in enumerate(reversed(y)):
            result[i + j] += Fraction(a) * Fraction(b)
    return list(range(1 - len(y), len(x))), [float(value) for value in result]


def _raises(expected, function, *args, **kwargs):
    try:
        function(*args, **kwargs)
    except expected:
        return True
    return False


def run_example():
    x, y = [1.0, 3.0, -2.0], [0.0, 0.0, 1.0, 3.0, -2.0]
    saved_x, saved_y = x.copy(), y.copy()
    result = cross_correlate(x, y)
    swapped = cross_correlate(y, x)
    expected_lags = [-4, -3, -2, -1, 0, 1, 2]
    expected_scores = [-2.0, -3.0, 14.0, -3.0, -2.0, 0.0, 0.0]
    x_programs = [p for count in range(1, 4) for p in itertools.product((-1.0, 0.0, 2.0), repeat=count)]
    y_programs = [p for count in range(1, 4) for p in itertools.product((-2.0, 1.0), repeat=count)]
    reference_checks = []
    for a in x_programs:
        for b in y_programs:
            actual = cross_correlate(a, b)
            lags, scores = reference_reversed_convolution(a, b)
            reference_checks.append(actual['lags'] == lags and actual['correlation_V2'] == scores)
    tied = cross_correlate([1.0], [2.0, 2.0])
    zeros = cross_correlate([0.0, 0.0], [0.0, 0.0])
    scaled = cross_correlate(x, [2 * value for value in y])
    self_result = cross_correlate(x, x)
    invalid_signals = ([], '123', iter([1.0]), [True], [float('nan')], [float('inf')],
                       ['1'], [None], [1j], [2**10000])
    checks = {
        'hand_full_lags_and_length': result['lags'] == expected_lags and len(result['correlation_V2']) == 7,
        'hand_asymmetric_correlation_values': result['correlation_V2'] == expected_scores,
        'delayed_second_signal_negative_lag': result['peak'] == {
            'index': 2, 'lag_samples': -2, 'correlation_V2': 14.0, 'unique': True, 'exact_ties': 1},
        'swap_reverses_scores_and_lags': swapped['correlation_V2'] == result['correlation_V2'][::-1]
            and swapped['lags'] == [-lag for lag in result['lags'][::-1]],
        'swapped_peak_has_positive_lag': swapped['peak']['lag_samples'] == 2,
        'delayed_first_impulse_positive_lag': cross_correlate([0, 1, 0], [1])['peak']['lag_samples'] == 1,
        'signed_maximum_is_not_absolute_maximum': cross_correlate([1], [-5, 2])['peak'] == {
            'index': 0, 'lag_samples': -1, 'correlation_V2': 2.0, 'unique': True, 'exact_ties': 1},
        'exact_peak_tie_smallest_lag_and_flag': tied['peak'] == {
            'index': 0, 'lag_samples': -1, 'correlation_V2': 2.0, 'unique': False, 'exact_ties': 2},
        'all_zero_peak_is_not_unique': zeros['correlation_V2'] == [0.0, 0.0, 0.0]
            and zeros['peak']['lag_samples'] == -1 and zeros['peak']['exact_ties'] == 3
            and zeros['peak']['unique'] is False,
        'hand_overlap_counts_include_zero_samples': result['overlap_samples'] == [1, 2, 3, 3, 3, 2, 1],
        'independent_rational_reversed_convolution': len(reference_checks) == 546 and all(reference_checks),
        'positive_gain_scales_raw_scores': scaled['correlation_V2'] == [2 * value for value in expected_scores]
            and scaled['peak']['lag_samples'] == -2 and scaled['peak']['correlation_V2'] == 28.0,
        'self_zero_lag_is_hand_energy': self_result['peak']['lag_samples'] == 0
            and self_result['correlation_V2'][self_result['lags'].index(0)] == 14.0,
        'singleton_signed_product_at_zero': cross_correlate([2], [-3]) == {
            'lags': [0], 'correlation_V2': [-6.0], 'overlap_samples': [1],
            'peak': {'index': 0, 'lag_samples': 0, 'correlation_V2': -6.0, 'unique': True, 'exact_ties': 1}},
        'sample_index_to_seconds_needs_rate': lag_seconds(-2, 1000) == -0.002
            and lag_seconds(2, 1000) == 0.002 and lag_seconds(0, 1000) == 0.0,
        'invalid_signals_and_voltage_units_rejected': all(
            _raises((TypeError, ValueError), cross_correlate, a, [1]) for a in invalid_signals)
            and all(_raises((TypeError, ValueError), cross_correlate, [1], b) for b in invalid_signals)
            and all(_raises(ValueError, cross_correlate, [1], [1], unit=unit) for unit in ('mV', b'V', None)),
        'rate_and_integer_lag_contract': all(
            _raises((TypeError, ValueError), lag_seconds, 1, rate)
            for rate in (True, 0, -1, float('nan'), float('inf'), '1000', 2**10000))
            and all(_raises(TypeError, lag_seconds, lag, 1000) for lag in (True, 1.0, None))
            and _raises(ValueError, lag_seconds, 1, 1000, unit='kHz')
            and _raises(ArithmeticError, lag_seconds, 2, math.ulp(0.0)),
        'overflow_rejected_without_input_mutation': _raises(ArithmeticError, cross_correlate, [1e308], [2])
            and _raises(ArithmeticError, cross_correlate, [1e308, 1e308], [1, 1])
            and x == saved_x and y == saved_y,
    }
    assert len(checks) == 18 and all(type(value) is bool and value for value in checks.values())
    return {'runtime': {'python': platform.python_version(), 'libraries': ['stdlib only']},
            'contract': {'formula_real': 'r_xy[k] = sum_n x[n] * y[n-k]',
                         'lags': '-(M-1) through N-1 inclusive', 'mode': 'full finite linear, zero outside support',
                         'input_unit': 'V for both', 'score_unit': 'V^2',
                         'peak': 'largest signed raw score; exact ties choose smallest lag and are flagged'},
            'metrics': {'x_V': x, 'y_V': y, 'result_xy': result, 'swapped_peak': swapped['peak'],
                        'peak_seconds_if_1000_Hz': lag_seconds(result['peak']['lag_samples'], 1000),
                        'gain_two_peak': scaled['peak'], 'tied_example_peak': tied['peak'],
                        'zero_example_peak': zeros['peak'], 'independent_reference_pairs': len(reference_checks)},
            'checks': checks,
            'limitations': ['No SciPy/NumPy execution, measured signal, FFT, circular wrap, device clock or synchronization test.',
                            'Raw sum: no demeaning, energy/overlap normalization or Pearson coefficient; gain/DC/overlap can affect scores.',
                            'Unique means one exact represented maximum only; not statistical confidence, causal direction or guaranteed physical delay.',
                            'Finite products/sums required; overflow rejected, float rounding/underflow and time conversion rounding remain possible.']}


if __name__ == '__main__':
    print(json.dumps(run_example(), ensure_ascii=False, indent=2))

آرایهٔ lag کنار آرایهٔ پاسخ ساخته می‌شود. API تابع correlation_lags نیز این نگاشت اندیس به جابه‌جایی را توضیح می‌دهد؛ مرجع علامت ما همان فرمول صفحهٔ correlate است. در مثال اصلی، قله در اندیس دو قرار دارد، ولی lag آن منفی دو است. برگرداندن عدد دو از argmax بدون نگاشت، جهت و مقدار گزارش را خراب می‌کند.

برای تبدیل اندیس به ثانیه، نرخ نمونه‌برداری معتبر و یکسان با واحد Hz و نمونه‌برداری یکنواخت را فرض می‌کنیم؛ تبدیل k/fs است. مقدار منفی دو در هزار نمونه بر ثانیه، منفی ۰٫۰۰۲ ثانیه می‌شود. این فقط تبدیلِ lag قراردادی است؛ مبدأ زمانی دو دستگاه، خطای ساعت و رانش بررسی نشده‌اند. جمع با fsum انجام می‌شود، اما گردکردن، کم‌ریزی و محدودیت نمایش عددی باقی می‌مانند.

۴. مرجع نامتقارن و هجده گروه بررسی

ورودی x برابر یک، سه و منفی دو ولت است. ورودی y دو صفر و سپس همان سه مقدار را دارد؛ الگوی آن در اندیس دو شروع می‌شود. lagها منفی چهار، منفی سه، منفی دو، منفی یک، صفر، یک و دو هستند. پاسخ دستی به‌ترتیب منفی دو، منفی سه، چهارده، منفی سه، منفی دو، صفر و صفر V² است. در lag منفی دو، جمع مربع‌های یک، سه و منفی دو برابر چهارده می‌شود.

{
  "runtime": {
    "python": "3.14.5",
    "libraries": [
      "stdlib only"
    ]
  },
  "contract": {
    "formula_real": "r_xy[k] = sum_n x[n] * y[n-k]",
    "lags": "-(M-1) through N-1 inclusive",
    "mode": "full finite linear, zero outside support",
    "input_unit": "V for both",
    "score_unit": "V^2",
    "peak": "largest signed raw score; exact ties choose smallest lag and are flagged"
  },
  "metrics": {
    "x_V": [
      1.0,
      3.0,
      -2.0
    ],
    "y_V": [
      0.0,
      0.0,
      1.0,
      3.0,
      -2.0
    ],
    "result_xy": {
      "lags": [
        -4,
        -3,
        -2,
        -1,
        0,
        1,
        2
      ],
      "correlation_V2": [
        -2.0,
        -3.0,
        14.0,
        -3.0,
        -2.0,
        0.0,
        0.0
      ],
      "overlap_samples": [
        1,
        2,
        3,
        3,
        3,
        2,
        1
      ],
      "peak": {
        "index": 2,
        "lag_samples": -2,
        "correlation_V2": 14.0,
        "unique": true,
        "exact_ties": 1
      }
    },
    "swapped_peak": {
      "index": 4,
      "lag_samples": 2,
      "correlation_V2": 14.0,
      "unique": true,
      "exact_ties": 1
    },
    "peak_seconds_if_1000_Hz": -0.002,
    "gain_two_peak": {
      "index": 2,
      "lag_samples": -2,
      "correlation_V2": 28.0,
      "unique": true,
      "exact_ties": 1
    },
    "tied_example_peak": {
      "index": 0,
      "lag_samples": -1,
      "correlation_V2": 2.0,
      "unique": false,
      "exact_ties": 2
    },
    "zero_example_peak": {
      "index": 0,
      "lag_samples": -1,
      "correlation_V2": 0.0,
      "unique": false,
      "exact_ties": 3
    },
    "independent_reference_pairs": 546
  },
  "checks": {
    "hand_full_lags_and_length": true,
    "hand_asymmetric_correlation_values": true,
    "delayed_second_signal_negative_lag": true,
    "swap_reverses_scores_and_lags": true,
    "swapped_peak_has_positive_lag": true,
    "delayed_first_impulse_positive_lag": true,
    "signed_maximum_is_not_absolute_maximum": true,
    "exact_peak_tie_smallest_lag_and_flag": true,
    "all_zero_peak_is_not_unique": true,
    "hand_overlap_counts_include_zero_samples": true,
    "independent_rational_reversed_convolution": true,
    "positive_gain_scales_raw_scores": true,
    "self_zero_lag_is_hand_energy": true,
    "singleton_signed_product_at_zero": true,
    "sample_index_to_seconds_needs_rate": true,
    "invalid_signals_and_voltage_units_rejected": true,
    "rate_and_integer_lag_contract": true,
    "overflow_rejected_without_input_mutation": true
  },
  "limitations": [
    "No SciPy/NumPy execution, measured signal, FFT, circular wrap, device clock or synchronization test.",
    "Raw sum: no demeaning, energy/overlap normalization or Pearson coefficient; gain/DC/overlap can affect scores.",
    "Unique means one exact represented maximum only; not statistical confidence, causal direction or guaranteed physical delay.",
    "Finite products/sums required; overflow rejected, float rounding/underflow and time conversion rounding remain possible."
  ]
}

قلهٔ این fixture یکتا و lag آن منفی دو است؛ با عوض کردن دو ورودی، پاسخ برعکس و lag قله مثبت دو می‌شود. این انتظار برای همین قلهٔ یکتا بررسی شده است؛ در قله‌های مساوی، سیاست انتخاب کوچک‌ترین lag لزوماً انتخاب‌های قرینه نمی‌دهد. دو برابر کردن y نیز قلهٔ خام را از چهارده به بیست‌وهشت می‌رساند؛ این افزایش، بهبود کیفیت تطبیق را اثبات نمی‌کند.

مرجع مستقل، هم‌نهشتی Fractionِ ورودی اول و ترتیب معکوس ورودی دوم را با پخش حاصل‌ضرب‌ها محاسبه می‌کند. این روش از حلقهٔ جمع‌آوری در lag تابع اصلی استفاده نمی‌کند. ۵۴۶ زوج دنبالهٔ محدود با آن مقایسه شده‌اند؛ یک گروه از هجده گروه بررسی‌اند. پاسخ دستی، ورودی تک‌عضوی، شمار هم‌پوشانی، قلهٔ مساوی و تفاوت بیشینهٔ علامت‌دار و قدرمطلق نیز جدا بررسی می‌شوند.

آزمون دامنهٔ FFT موضوع متفاوتی را می‌سنجد؛ اینجا محاسبه مستقیم است و FFT نداریم. برای ثبت فرض‌ها نیز آزمایش تکرارپذیر سیگنال مکمل است. پیش از انتشار، فرمول، ترتیب ورودی، واحد و پاسخ طلایی را با هم مرور کنیم. عوض‌کردن علامت خروجی فقط برای موافق‌شدن با انتظار مبهم، جای اصلاح قرارداد را نمی‌گیرد.

۵. چهار پرسش پاسخ‌دار دربارهٔ علامت و تفسیر

پرسش اول: چرا الگوی شروع‌شده در اندیس دوِ y، lag منفی دو می‌دهد؟ پاسخ: در فرمول x[n]y[n−k] با k منفی دو، مقدار y[n+2] با x[n] جفت می‌شود. جهت از همین قرارداد می‌آید؛ با جابه‌جایی ورودی‌ها قلهٔ یکتای مثال مثبت دو است.

پرسش دوم: اندیس قلهٔ دو یعنی دو نمونه lag؟ پاسخ: خیر؛ باید lags[2] را بخوانیم که اینجا منفی دو است. فقط با نرخ فرضی هزار هرتز، lag زمانیِ قراردادی منفی دو میلی‌ثانیه می‌شود.

پرسش سوم: پاسخ چهارده، ضریب هم‌بستگی نامعتبر است؟ پاسخ: خیر؛ این جمع خام با واحد V² است، نه ضریب نرمال‌شده. چهارده جمع مربع نمونه‌های انتخابی است؛ آن را انرژی فیزیکی بر حسب ژول نمی‌نامیم.

پرسش چهارم: قلهٔ یکتا یا هجده بررسی موفق، هم‌زمانی دستگاه‌ها را ثابت می‌کند؟ پاسخ: خیر؛ فقط محاسبه و قرارداد داده‌های ساختگی بررسی شده‌اند. ساعت، نویز واقعی و علت مشترک اندازه‌گیری نشده‌اند. منابع در ۲۴ مهر ۱۴۰۵ خوانده شدند و نتیجه به تأخیر فیزیکی تعمیم داده نمی‌شود.

منابع و مطالعهٔ بیشتر

• SciPy correlate؛ فرمول، گسترهٔ lag و امتداد صفر
• SciPy correlation_lags؛ نگاشت اندیس خروجی به lag
• NumPy correlate؛ تفاوت قراردادهای علامت و جابه‌جایی ورودی
• Python 3.14؛ جمع fsum و کنترل finite

بررسی منابع: ۲۴ مهر ۱۴۰۵. تحلیل‌ها و پیشنهادهای اجرایی، جمع‌بندی نویسنده‌اند.