ذاكرة التخزين المؤقت المجمعة لـ Parquet: كيف تُسرّع الاختبارات الرجعية متعددة الأُطُر الزمنية بمئات المرات

تستخدم الاستراتيجية متعددة الأُطُر الزمنية عدة أُطُر زمنية في وقت واحد: اليومي يحدد اتجاه الترند، والساعي يحدد نقاط الدخول، والخمس دقائق يحدد توقيت التنفيذ بدقة. يحتاج كل إطار زمني إلى مؤشراته الخاصة: المتوسطات المتحركة، والمذبذبات، والمستويات.

لاختبار رجعي واحد، كل شيء بسيط — أعد حساب الأُطُر الزمنية من بيانات الدقيقة، واحسب المؤشرات، وشغّل الاستراتيجية. لكن أثناء التحسين المكثف — عندما تحتاج لاختبار آلاف من مجموعات المعاملات — تصبح إعادة حساب الأُطُر الزمنية والمؤشرات في كل تكرار عنق زجاجة. مرور واحد عبر بيانات الدقيقة على مدار عامين يعني معالجة أكثر من مليون شريط، وتكرار ذلك ألف مرة هو هدر.

الحل: حساب كل شيء مرة واحدة وتخزينه مؤقتاً في ملف parquet.

المشكلة: حسابات متكررة أثناء التحسين

خط أنابيب اختبار رجعي نموذجي متعدد الأُطُر الزمنية:

for params in parameter_grid:
    df_1m = load_candles("ETHUSDT", "1m", start, end)

    df_5m = resample_ohlcv(df_1m, "5m")
    df_1h = resample_ohlcv(df_1m, "1h")
    df_4h = resample_ohlcv(df_1m, "4h")
    df_1d = resample_ohlcv(df_1m, "D")

    ma_1h = compute_ma(df_1h["close"], length=params["ma_1h_len"])
    ma_4h = compute_ma(df_4h["close"], length=params["ma_4h_len"])
    ma_1d = compute_ma(df_1d["close"], length=params["ma_1d_len"])

    result = run_strategy(df_1m, ma_1h, ma_4h, ma_1d, params)

في كل تكرار، يتم إعادة حساب الخطوات 1-3 رغم أن البيانات هي نفسها. فقط معاملات عتبة الاستراتيجية تتغير (الخطوة 4). إنه مثل إعادة بناء منزل كامل في كل مرة تريد فيها فقط تجربة لون حائط مختلف.

الفكرة: حساب مرة واحدة، حفظ، إعادة استخدام مرات عديدة

الملاحظة الأساسية: الأُطُر الزمنية والمؤشرات تعتمد فقط على بيانات الدقيقة ومعاملات المؤشرات، وليس على معاملات الاستراتيجية. إذا ثبّتنا مجموعة المؤشرات المطلوبة، يمكننا حسابها مرة واحدة وحفظها.

المخطط:

الخطوة 1 (مرة واحدة):
  شموع الدقيقة -> إعادة تشكيل الأُطُر الزمنية -> حساب المؤشرات -> ملف Parquet

الخطوة 2 (مرات عديدة):
  ملف Parquet -> استراتيجية بمعاملات مختلفة -> النتيجة

محاكاة الأُطُر الزمنية من شموع الدقيقة

لدينا أرشيف كامل من شموع الدقيقة. منه يمكننا إعادة إنتاج أي إطار زمني أعلى بدقة. لكن هناك فارق دقيق: مع resample القياسي، نحصل على صف واحد لكل فترة (صف واحد لكل ساعة، صف لكل 4 ساعات، إلخ). هذا لا يصلح للاختبار الرجعي دقيقة بدقيقة — نحتاج لمعرفة قيمة المؤشر عند كل دقيقة.

لذلك، نقوم بـمحاكاة قيم الإطار الزمني الأعلى لكل شمعة دقيقة، مُحاكين كيف يرى البوت البيانات في الوقت الفعلي:

1. يتلقى البوت شمعة الدقيقة التالية
2. يُحدّث الشريط الحالي (غير المكتمل) للإطار الزمني الأعلى — يعيد حساب High وLow وClose وVolume
3. يعيد حساب المؤشر عبر جميع الأشرطة المكتملة بالإضافة إلى الشريط الجزئي الحالي
4. عندما تنتهي الفترة — يتم إنهاء الشريط ويبدأ شريط جديد

يضمن هذا النهج أن الاختبار الرجعي يرى نفس البيانات بالضبط التي يراها البوت في الوقت الفعلي. لا نظر إلى المستقبل — كل شمعة دقيقة تُعالج بصرامة بالبيانات التي كانت ستكون متاحة في تلك اللحظة.

class RunningCandleBuffer:
    """
    Emulates real-time updates of a higher timeframe bar
    using 1-minute candles.
    """
    def __init__(self, period_seconds: int):
        self.period = period_seconds  # 86400 for Daily, 3600 for 1h
        self.closed_bars = []
        self.current_bar = None

    def update(self, timestamp, open_, high, low, close, volume):
        bar_start = self._align_to_period(timestamp)

        if self.current_bar is None or bar_start != self.current_bar['start']:
            if self.current_bar is not None:
                self.closed_bars.append(self.current_bar)
            self.current_bar = {
                'start': bar_start,
                'open': open_, 'high': high,
                'low': low, 'close': close,
                'volume': volume,
            }
        else:
            self.current_bar['high'] = max(self.current_bar['high'], high)
            self.current_bar['low'] = min(self.current_bar['low'], low)
            self.current_bar['close'] = close
            self.current_bar['volume'] += volume

        return self.closed_bars + [self.current_bar]

يتم إنشاء RunningCandleBuffer منفصل لكل إطار زمني أعلى. عند كل شمعة دقيقة، يتم تحديث جميع المخازن المؤقتة، مما يعطينا الحالة الحالية لكل إطار زمني — كما لو كان البوت يعمل في الوقت الفعلي.

هيكل ذاكرة التخزين المؤقت Parquet

نتيجة الحساب المسبق هي ملف parquet واحد حيث يتوافق كل صف مع شمعة دقيقة واحدة، والأعمدة تحتوي على:

timestamp              — الطابع الزمني لشمعة الدقيقة
open, high, low,       — OHLCV لشمعة الدقيقة
close, volume

close_5m               — Close للشمعة المحاكاة 5m عند هذه اللحظة
close_1h               — Close للشمعة المحاكاة 1h
close_4h               — Close للشمعة المحاكاة 4h
close_1d               — Close للشمعة اليومية المحاكاة

ma_20_1h               — MA(20) على 1h، معاد حسابه عند هذه الدقيقة
ma_50_1h               — MA(50) على 1h
ma_20_4h               — MA(20) على 4h
ma_50_4h               — MA(50) على 4h
ma_6_1d                — MA(6) على اليومي
ma_12_1d               — MA(12) على اليومي

cross_ma_1h            — إشارة تقاطع MA على 1h ('buy'/'sell'/None)
cross_ma_4h            — إشارة تقاطع MA على 4h
cross_ma_1d            — إشارة تقاطع MA على اليومي

separation_1h          — تباعد MA بالنسبة المئوية على 1h
separation_4h          — تباعد MA بالنسبة المئوية على 4h
separation_1d          — تباعد MA بالنسبة المئوية على اليومي

كل قيمة تعكس الحالة الفعلية للمؤشر عند لحظة شمعة الدقيقة المقابلة — مع الأخذ بعين الاعتبار الأشرطة غير المكتملة للأُطُر الزمنية الأعلى.

الحساب المسبق: بناء ذاكرة التخزين المؤقت

def precompute_cache(
    df_1m: pd.DataFrame,
    timeframes: dict[str, int],   # {"5m": 300, "1h": 3600, "4h": 14400, "D": 86400}
    indicators: dict,              # {"ma_20": 20, "ma_50": 50}
) -> pd.DataFrame:
    """
    Single pass through all minute candles.
    Returns a DataFrame with emulated timeframes and indicators.
    """
    buffers = {tf: RunningCandleBuffer(secs) for tf, secs in timeframes.items()}

    n = len(df_1m)
    result = {}

    for tf_name, buf in buffers.items():
        closes = np.zeros(n)
        ma_values = {name: np.full(n, np.nan) for name in indicators}

        for i in range(n):
            row = df_1m.iloc[i]
            bars = buf.update(
                df_1m.index[i],
                row['open'], row['high'], row['low'], row['close'], row['volume']
            )

            all_closes = [b['close'] for b in bars]
            closes[i] = all_closes[-1]

            for ind_name, length in indicators.items():
                if len(all_closes) >= length:
                    ma_values[ind_name][i] = np.mean(all_closes[-length:])

        result[f'close_{tf_name}'] = closes
        for ind_name in indicators:
            result[f'{ind_name}_{tf_name}'] = ma_values[ind_name]

    cache_df = pd.DataFrame(result, index=df_1m.index)
    cache_df = pd.concat([df_1m[['open', 'high', 'low', 'close', 'volume']], cache_df], axis=1)

    return cache_df

cache = precompute_cache(
    df_1m,
    timeframes={"5m": 300, "1h": 3600, "4h": 14400, "D": 86400},
    indicators={"ma_20": 20, "ma_50": 50, "ma_6": 6, "ma_12": 12},
)

cache.to_parquet("cache_ETHUSDT_2024_2026.parquet")

استخدام ذاكرة التخزين المؤقت أثناء التحسين

الآن يبدو التحسين هكذا:

cache = pd.read_parquet("cache_ETHUSDT_2024_2026.parquet")

for params in parameter_grid:
    result = run_strategy(cache, params)

تعمل الاستراتيجية مع الأعمدة المبنية مسبقاً — لا مرور متكرر عبر مليون شريط، ولا إعادة حساب MA، ولا محاكاة أُطُر زمنية. فقط قراءة من DataFrame والتحقق من شروط الدخول/الخروج.

لماذا Parquet

Parquet هو تنسيق تخزين بيانات عمودي، مثالي لهذه المهمة:

• الضغط. يضغط Parquet البيانات الرقمية 5-10 أضعاف. ذاكرة تخزين مؤقت من 1.1 مليون صف مع 30 عمود تأخذ ~50 ميغابايت بدلاً من ~500 ميغابايت في CSV.
• القراءة العمودية. إذا كانت الاستراتيجية تستخدم فقط ma_20_4h وma_50_4h، يقرأ parquet تلك الأعمدة فقط متخطياً الباقي.
• حفظ الأنواع. أنواع البيانات (float64, int64, string) تُحفظ بدون خسارة — لا حاجة لتحليل سلاسل نصية عند التحميل.
• سرعة القراءة. تحميل parquet في pandas يستغرق عشرات الميلي ثوانٍ، أسرع بمرتبة من CSV.

توسيع ذاكرة التخزين المؤقت: إضافة مؤشرات جديدة

إذا احتاجت الاستراتيجية لمؤشر جديد (RSI أو MACD أو بولينجر باندز)، ببساطة:

1. أعد حساب المؤشر الجديد فقط من نفس بيانات الدقيقة
2. أضف الأعمدة إلى ملف parquet الحالي
3. جميع الأعمدة المحسوبة مسبقاً تبقى كما هي

cache = pd.read_parquet("cache_ETHUSDT_2024_2026.parquet")

rsi_cols = compute_rsi_for_timeframes(df_1m, timeframes, length=14)

cache = pd.concat([cache, rsi_cols], axis=1)
cache.to_parquet("cache_ETHUSDT_2024_2026.parquet")

الملخص: مقارنة الأساليب

الخلاصة

نهج ذاكرة التخزين المؤقت المجمعة لـ Parquet يحل مشكلتين في وقت واحد:

1. الصحة. محاكاة الأُطُر الزمنية من شموع الدقيقة عبر RunningCandleBuffer تضمن أن الاختبار الرجعي يرى نفس البيانات التي يراها البوت في الوقت الفعلي — لا نظر إلى المستقبل ولا تأخيرات اصطناعية.

2. السرعة. الأُطُر الزمنية والمؤشرات المحسوبة مسبقاً تتيح اختبار آلاف مجموعات المعاملات في دقائق بدلاً من أيام.

الفكرة بسيطة: حساب مرة واحدة — إعادة استخدام مرات عديدة. شموع الدقيقة هي البيانات المصدر. كل شيء آخر مشتق ويمكن حسابه مسبقاً وتخزينه مؤقتاً. يجعل Parquet هذه الذاكرة المؤقتة مضغوطة وسريعة ومريحة.

لمزيد من المعلومات حول كيفية تحسين دقة محاكاة التنفيذ بالحفر التكيفي من الدقائق إلى الثواني والميلي ثوانٍ، انظر مقال الحفر التكيفي: اختبار رجعي بدقة متغيرة.

روابط مفيدة

1. Apache Parquet — data storage format
2. pandas — working with parquet
3. Lopez de Prado — Advances in Financial Machine Learning
4. Ernest Chan — Quantitative Trading

Citation

@article{soloviov2026parquetcache,
  author = {Soloviov, Eugen},
  title = {Aggregated Parquet Cache: How to Speed Up Multi-Timeframe Backtests by Hundreds of Times},
  year = {2026},
  url = {https://marketmaker.cc/ru/blog/post/parquet-cache-multitimeframe-backtest},
  description = {How to precompute timeframes and indicators from minute candles, save them to parquet, and use them for mass strategy testing without redundant recalculations.}
}