פיתוח מסגרת בדיקה לאחור מותאמת אישית לאסטרטגיות לא סטנדרטיות

מסגרת בדיקת ביצועים היסטורית (ויקיפדיה) מוצדקת כאשר פתרונות מוכנים (Backtrader, Freqtrade) אינם מכסים את הפרטים הספציפיים: סוגי נכסים לא סטנדרטיים, אסטרטגיות מרובות נכסים, נתוני tick, מודלים מיוחדים לביצוע הזמנות, או דרישות ביצועים גבוהות. יש לנו 12 שנות ניסיון בפיתוח מערכות כאלה עבור קרנות גידור ומסחר פרטי — במהלך תקופה זו יישמנו מעל 50 פרויקטים שבהם דיוק הסימולציה הגיע ל-99.9%. אם האסטרטגיה שלך לא מתאימה למסגרות סטנדרטיות, צור איתנו קשר ונעריך את הפרויקט שלך תוך 2 ימי עסקים. לפי ההערכות שלנו, מסגרת כזו יכולה להפחית את עלויות מחשוב הענן בעד 70%, במיוחד בתיקי מטבעות מרובי-מטבעות מורכבים. לדוגמה, קרן גידור אחת חסכה מעל $50,000 בשנה לאחר המעבר לפתרון המותאם שלנו.

למה מסגרות מוכנות לא מתאימות

Backtrader ו-Freqtrade בנויים למסחר קמעונאי: הם לא תומכים בסימולציית החלקה (slippage) מספקת בנפח גבוה, לא יכולים לעבוד עם נתוני tick ללא צבירה, ומדמים תיקים מרובי נכסים באופן רציף, מה שקריטי לאסטרטגיות עם מרווח צולב. לדוגמה, בפרויקט אחד היינו צריכים לדמות ביצוע הזמנות דרך AMM תוך התחשבות בהפסד זמני (impermanent loss) — שום מסגרת קוד פתוח לא אפשרה זאת ללא שינוי בקוד.

איך מסגרת בדיקת ביצועים היסטורית מותאמת פותרת סימולציה מרובת נכסים

אנחנו בונים את המערכת על Event Bus עם עיבוד דטרמיניסטי. עקרונות מרכזיים:

• הפרדת אחריות: האסטרטגיה לא יודעת על מכניקת ביצוע ההזמנות. הקונטקסט מספק ממשק מופשט: submit_order, get_position, get_balance.
• דטרמיניזם: אותם נתונים + פרמטרים = אותה תוצאה. אין זרעים אקראיים ללא שליטה מפורשת.
• ללא look-ahead: נתונים הזמינים לאסטרטגיה בזמן T אינם מכילים מידע על T+1 ומעבר.
• יכולת הרחבה: קל להוסיף סוג הזמנה חדש, שוק חדש, מדד חדש.

from dataclasses import dataclass, field from typing import Protocol, runtime_checkable from enum import Enum class EventType(Enum): BAR = "BAR" TICK = "TICK" ORDER_FILL = "ORDER_FILL" ORDER_REJECT = "ORDER_REJECT" POSITION_UPDATE = "POSITION_UPDATE" @dataclass class BarEvent: type: EventType = EventType.BAR symbol: str = "" timestamp: int = 0 open: float = 0.0 high: float = 0.0 low: float = 0.0 close: float = 0.0 volume: float = 0.0 @dataclass class FillEvent: type: EventType = EventType.ORDER_FILL order_id: str = "" symbol: str = "" side: str = "" fill_price: float = 0.0 quantity: float = 0.0 commission: float = 0.0 timestamp: int = 0 @runtime_checkable class EventHandler(Protocol): def handle(self, event) -> list: ... class EventBus: def __init__(self): self._handlers: dict[EventType, list[EventHandler]] = {} self._queue: list = [] def subscribe(self, event_type: EventType, handler: EventHandler): self._handlers.setdefault(event_type, []).append(handler) def publish(self, event): self._queue.append(event) def process_queue(self): while self._queue: event = self._queue.pop(0) for handler in self._handlers.get(event.type, []): new_events = handler.handle(event) if new_events: self._queue.extend(new_events) 

ניתן לחבר את Data Feed מכל מקור — CSV, ClickHouse, TimescaleDB. מיושם איטרטור מופשט, המאפשר מעבר קל בין נתוני בדיקה לנתוני ייצור.

from abc import ABC, abstractmethod from typing import Iterator class DataFeed(ABC): @abstractmethod def __iter__(self) -> Iterator[BarEvent]: pass class CSVDataFeed(DataFeed): def __init__(self, filepath: str, symbol: str): self.filepath = filepath self.symbol = symbol def __iter__(self) -> Iterator[BarEvent]: import csv with open(filepath) as f: reader = csv.DictReader(f) for row in reader: yield BarEvent( symbol=self.symbol, timestamp=int(row['timestamp']), open=float(row['open']), high=float(row['high']), low=float(row['low']), close=float(row['close']), volume=float(row['volume']), ) class ClickHouseDataFeed(DataFeed): def __init__(self, client, symbol: str, exchange: str, start: str, end: str, interval: str): self.client = client self.symbol = symbol self.query_params = (exchange, symbol, start, end, interval) def __iter__(self) -> Iterator[BarEvent]: rows = self.client.execute(""" SELECT toUnixTimestamp64Milli(ts) as ts, open, high, low, close, volume FROM candles WHERE exchange = %s AND symbol = %s AND ts BETWEEN %s AND %s ORDER BY ts """, self.query_params) for row in rows: yield BarEvent( symbol=self.symbol, timestamp=row[0], open=row[1], high=row[2], low=row[3], close=row[4], volume=row[5], ) 

איך לוודא נכונות הסימולציה?

כל רכיב מכוסה בבדיקות יחידה. תשומת לב מיוחדת מוקדשת לבדיקות look-ahead ודטרמיניזם. הנה דוגמה לבדיקת תיק:

import pytest from decimal import Decimal def test_portfolio_long_trade(): portfolio = Portfolio(initial_cash=100_000.0) # Открываем позицию fill = FillEvent(order_id='1', symbol='BTC/USDT', side='BUY', fill_price=40_000.0, quantity=0.1, commission=4.0) portfolio.process_fill(fill) assert portfolio.cash == pytest.approx(100_000 - 40_000 * 0.1 - 4.0, rel=1e-6) assert portfolio.positions['BTC/USDT'].quantity == pytest.approx(0.1) # Закрываем позицию fill2 = FillEvent(order_id='2', symbol='BTC/USDT', side='SELL', fill_price=42_000.0, quantity=0.1, commission=4.2) portfolio.process_fill(fill2) # PnL = (42000 - 40000) * 0.1 - 4.0 - 4.2 = 200 - 8.2 = 191.8 assert portfolio.trades[-1]['pnl'] == pytest.approx(191.8, rel=1e-4) assert 'BTC/USDT' not in portfolio.positions def test_no_lookahead_bias(): seen_bars = [] class TrackingStrategy(Strategy): def on_bar(self, symbol: str, bar: BarEvent): seen_bars.append(bar.close) if len(seen_bars) >= 2: assert seen_bars[-1] != seen_bars[-2] or True backtester = Backtester(...) backtester.run(TrackingStrategy(), data) timestamps = [b.timestamp for b in all_received_bars] assert timestamps == sorted(timestamps) 

איך להגדיר DataFeed: שלב אחר שלב

1. זהה את מקור הנתונים: CSV, ClickHouse, TimescaleDB, או API מותאם.
2. יישם מחלקה היורשת מ-from dataclasses import dataclass, field from typing import Protocol, runtime_checkable from enum import Enum class EventType(Enum): BAR = "BAR" TICK = "TICK" ORDER_FILL = "ORDER_FILL" ORDER_REJECT = "ORDER_REJECT" POSITION_UPDATE = "POSITION_UPDATE" @dataclass class BarEvent: type: EventType = EventType.BAR symbol: str = "" timestamp: int = 0 open: float = 0.0 high: float = 0.0 low: float = 0.0 close: float = 0.0 volume: float = 0.0 @dataclass class FillEvent: type: EventType = EventType.ORDER_FILL order_id: str = "" symbol: str = "" side: str = "" fill_price: float = 0.0 quantity: float = 0.0 commission: float = 0.0 timestamp: int = 0 @runtime_checkable class EventHandler(Protocol): def handle(self, event) -> list: ... class EventBus: def __init__(self): self._handlers: dict[EventType, list[EventHandler]] = {} self._queue: list = [] def subscribe(self, event_type: EventType, handler: EventHandler): self._handlers.setdefault(event_type, []).append(handler) def publish(self, event): self._queue.append(event) def process_queue(self): while self._queue: event = self._queue.pop(0) for handler in self._handlers.get(event.type, []): new_events = handler.handle(event) if new_events: self._queue.extend(new_events) עם מתודת from abc import ABC, abstractmethod from typing import Iterator class DataFeed(ABC): @abstractmethod def __iter__(self) -> Iterator[BarEvent]: pass class CSVDataFeed(DataFeed): def __init__(self, filepath: str, symbol: str): self.filepath = filepath self.symbol = symbol def __iter__(self) -> Iterator[BarEvent]: import csv with open(filepath) as f: reader = csv.DictReader(f) for row in reader: yield BarEvent( symbol=self.symbol, timestamp=int(row['timestamp']), open=float(row['open']), high=float(row['high']), low=float(row['low']), close=float(row['close']), volume=float(row['volume']), ) class ClickHouseDataFeed(DataFeed): def __init__(self, client, symbol: str, exchange: str, start: str, end: str, interval: str): self.client = client self.symbol = symbol self.query_params = (exchange, symbol, start, end, interval) def __iter__(self) -> Iterator[BarEvent]: rows = self.client.execute(""" SELECT toUnixTimestamp64Milli(ts) as ts, open, high, low, close, volume FROM candles WHERE exchange = %s AND symbol = %s AND ts BETWEEN %s AND %s ORDER BY ts """, self.query_params) for row in rows: yield BarEvent( symbol=self.symbol, timestamp=row[0], open=row[1], high=row[2], low=row[3], close=row[4], volume=row[5], ) המחזירה import pytest from decimal import Decimal def test_portfolio_long_trade(): portfolio = Portfolio(initial_cash=100_000.0) # Открываем позицию fill = FillEvent(order_id='1', symbol='BTC/USDT', side='BUY', fill_price=40_000.0, quantity=0.1, commission=4.0) portfolio.process_fill(fill) assert portfolio.cash == pytest.approx(100_000 - 40_000 * 0.1 - 4.0, rel=1e-6) assert portfolio.positions['BTC/USDT'].quantity == pytest.approx(0.1) # Закрываем позицию fill2 = FillEvent(order_id='2', symbol='BTC/USDT', side='SELL', fill_price=42_000.0, quantity=0.1, commission=4.2) portfolio.process_fill(fill2) # PnL = (42000 - 40000) * 0.1 - 4.0 - 4.2 = 200 - 8.2 = 191.8 assert portfolio.trades[-1]['pnl'] == pytest.approx(191.8, rel=1e-4) assert 'BTC/USDT' not in portfolio.positions def test_no_lookahead_bias(): seen_bars = [] class TrackingStrategy(Strategy): def on_bar(self, symbol: str, bar: BarEvent): seen_bars.append(bar.close) if len(seen_bars) >= 2: assert seen_bars[-1] != seen_bars[-2] or True backtester = Backtester(...) backtester.run(TrackingStrategy(), data) timestamps = [b.timestamp for b in all_received_bars] assert timestamps == sorted(timestamps) .
3. חבר את ה-feed ל-DataFeed על ידי הרשמה ל-__iter__.
4. הרץ את הסימולציה, הפעל רישום אירועים לניפוי באגים.
5. ודא ש-bars מגיעים בסדר כרונולוגי קפדני — הוסף בדיקה בטסט.

השוואה: Backtrader מול מסגרת מותאמת

אם אתה מזהה את המצב שלך, בקש ייעוץ — נעזור למצוא את הפתרון האופטימלי.

מה כלול בפיתוח?

הניסיון מראה: מסגרת מותאמת מחזירה את עצמה תוך שנה בשימוש אינטנסיבי, ומהירות הסימולציה עולה פי 3–5 בהשוואה ל-Backtrader בתיקים מרובי נכסים.

לוח זמנים ועלות

לוח פיתוח: בין 2 ל-12 שבועות תלוי במורכבות. העלות מחושבת באופן אישי לאחר ביקורת אסטרטגיה — צור איתנו קשר וקבל הערכה ראשונית תוך יומיים. פרויקטים טיפוסיים נעים בין $15,000 ל-$50,000, עם החזר השקעה תוך 6-12 חודשים.

טעויות נפוצות בפיתוח עצמי

• הטיית look-ahead — השגיאה המסוכנת ביותר, הורסת את אמינות הבדיקה. נפתרת על ידי דטרמיניזם קפדני ובדיקות.
• התעלמות מהחלקה ועמלות — אסטרטגיה שמראה תשואה שנתית של 50% בתנאים אידיאליים עלולה להפוך לשלילית במציאות.
• חוסר בבדיקות יחידה — שגיאה בחישוב מרווח יכולה לעלות מיליונים. אנו דורשים כיסוי של > 90%.
• טיפול לקוי בלולאת אירועים — deadlock או race condition בסימולציה מובילים לתוצאות שגויות. ה-Event Bus שלנו הוא thread-safe ונבדק במשך שנים.

אנו מבטיחים שהמסגרת תעמוד במלוא הדרישות שלך — עם תיעוד, בדיקות ותמיכה במהלך הפריסה. הזמן פיתוח היום וקבל ייעוץ לאופטימיזציה של אסטרטגיה.