פיתוח מנוע בדיקת אסטרטגיות מותאם אישית ב-Python

אסטרטגיית מסחר מראה רווח על נתונים היסטוריים ב-Backtrader. אבל בייצור, החלקה, עמלות, ביצועים חלקיים ועיכוב בביצוע הורגים את התוצאות. כלים סטנדרטיים נותנים רק תמונה משוערת, ומחיר הטעות הוא הפסדים אמיתיים. אנו מתכננים מנועי בדיקת אסטרטגיה (Backtesting) מותאמים אישית ב-Python שמתחשבים במציאות השוק ומייצרים מדדים שאפשר לסמוך עליהם. הגישה שלנו מאפשרת שילוב של כל לוגיקת ביצוע, תוכניות עמלות לא סטנדרטיות, ואינטגרציה עם הנתונים שלך. במשך 8 שנים, תכננו 15+ מנועים כאלה עבור קרנות וחברות מסחר פרטיות. הניסיון מראה: מודל מדויק של החלקה וביצועים חלקיים מפחית את הפער מול מסחר חי ל-5%. קבלו ייעוץ—נעריך את הפרויקט שלכם תוך 2 ימי עסקים.

איך בדיקת אסטרטגיה ב-Python עוזרת למנוע הפסדים?

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

למה פתרונות מוכנים לא עומדים בציפיות?

Backtrader, Freqtrade, Zipline—פרויקטים בוגרים, אבל הם כופים ארכיטקטורה משלהם. כשצריך התאמת הזמנות לא סטנדרטית (מכירה פומבית, בריכה אפלה), עמלות מורכבות (רב-מטבעיות, הנחות נפח), או מקור נתונים קנייני—או שפורצים את הספרייה או בונים מנוע משלך. האחרון נותן שליטה וביצועים. מנוע מותאם אישית מדגמן החלקה פי 5 יותר מדויק מ-Backtrader באמצעות מקדמים דינמיים. חיסכון מהפסדי מודל לא מדויק יכול להגיע ל-30%.

איך אנו מתכננים את הארכיטקטורה

בסיס—מחלקה אבסטרקטית נקייה Strategy עם מתודת on_bar. אנו משתמשים ב-dataclasses עבור Bar, Order, Position. דוגמת שלד:

from abc import ABC, abstractmethod from dataclasses import dataclass, field from decimal import Decimal from typing import Optional import pandas as pd @dataclass class Bar: timestamp: pd.Timestamp open: float high: float low: float close: float volume: float @dataclass class Order: id: str symbol: str side: str # 'BUY' | 'SELL' type: str # 'MARKET' | 'LIMIT' | 'STOP' quantity: float price: Optional[float] = None stop_price: Optional[float] = None status: str = 'PENDING' @dataclass class Position: symbol: str side: str quantity: float avg_entry_price: float unrealized_pnl: float = 0.0 realized_pnl: float = 0.0 class Strategy(ABC): def __init__(self, context: 'BacktestContext'): self.ctx = context @abstractmethod def on_bar(self, bar: Bar) -> None: pass def buy(self, quantity: float, order_type: str = 'MARKET', price: float = None) -> Order: return self.ctx.submit_order(Order( id=self.ctx.generate_id(), symbol=self.ctx.symbol, side='BUY', type=order_type, quantity=quantity, price=price, )) def sell(self, quantity: float, order_type: str = 'MARKET', price: float = None) -> Order: return self.ctx.submit_order(Order( id=self.ctx.generate_id(), symbol=self.ctx.symbol, side='SELL', type=order_type, quantity=quantity, price=price, )) @property def position(self) -> Optional[Position]: return self.ctx.get_position(self.ctx.symbol) @property def cash(self) -> float: return self.ctx.portfolio.cash 

ניהול תיק וחשבונות פוזיציות

אנו מתחזקים היסטוריית עסקאות ועקומת הון. חשבונאות FIFO וחישוב אוטומטי של רווח/הפסד ממומש ולא ממומש.

class Portfolio: def __init__(self, initial_cash: float): self.initial_cash = initial_cash self.cash = initial_cash self.positions: dict[str, Position] = {} self.trades: list[dict] = [] self.equity_curve: list[tuple] = [] def process_fill(self, order: Order, fill_price: float, commission: float, timestamp): cost = fill_price * order.quantity if order.side == 'BUY': self.cash -= (cost + commission) symbol = order.symbol if symbol in self.positions: pos = self.positions[symbol] total_qty = pos.quantity + order.quantity pos.avg_entry_price = ( pos.avg_entry_price * pos.quantity + fill_price * order.quantity ) / total_qty pos.quantity = total_qty else: self.positions[symbol] = Position( symbol=symbol, side='LONG', quantity=order.quantity, avg_entry_price=fill_price, ) elif order.side == 'SELL': self.cash += (cost - commission) pos = self.positions.get(order.symbol) if pos: realized_pnl = (fill_price - pos.avg_entry_price) * order.quantity - commission pos.quantity -= order.quantity pos.realized_pnl += realized_pnl self.trades.append({ 'timestamp': timestamp, 'symbol': order.symbol, 'entry': pos.avg_entry_price, 'exit': fill_price, 'quantity': order.quantity, 'pnl': realized_pnl, }) if pos.quantity <= 0: del self.positions[order.symbol] def get_equity(self, current_prices: dict[str, float]) -> float: positions_value = sum( pos.quantity * current_prices.get(symbol, pos.avg_entry_price) for symbol, pos in self.positions.items() ) return self.cash + positions_value 

איך ליישם ביצוע ריאלי?

תכונה מרכזית—from abc import ABC, abstractmethod from dataclasses import dataclass, field from decimal import Decimal from typing import Optional import pandas as pd @dataclass class Bar: timestamp: pd.Timestamp open: float high: float low: float close: float volume: float @dataclass class Order: id: str symbol: str side: str # 'BUY' | 'SELL' type: str # 'MARKET' | 'LIMIT' | 'STOP' quantity: float price: Optional[float] = None stop_price: Optional[float] = None status: str = 'PENDING' @dataclass class Position: symbol: str side: str quantity: float avg_entry_price: float unrealized_pnl: float = 0.0 realized_pnl: float = 0.0 class Strategy(ABC): def __init__(self, context: 'BacktestContext'): self.ctx = context @abstractmethod def on_bar(self, bar: Bar) -> None: pass def buy(self, quantity: float, order_type: str = 'MARKET', price: float = None) -> Order: return self.ctx.submit_order(Order( id=self.ctx.generate_id(), symbol=self.ctx.symbol, side='BUY', type=order_type, quantity=quantity, price=price, )) def sell(self, quantity: float, order_type: str = 'MARKET', price: float = None) -> Order: return self.ctx.submit_order(Order( id=self.ctx.generate_id(), symbol=self.ctx.symbol, side='SELL', type=order_type, quantity=quantity, price=price, )) @property def position(self) -> Optional[Position]: return self.ctx.get_position(self.ctx.symbol) @property def cash(self) -> float: return self.ctx.portfolio.cash . היא מדגמנת החלקה, עמלות וביצועים חלקיים. עבור הזמנות שוק, היא משתמשת במחיר הפתיחה של הבר הבא עם החלקה. הזמנות לימיט בודקות אם הבר הגיע לרמה שצוינה. הזמנות עצירה מופעלות עם החלקה נוספת—מדמות פערי מחיר. לדוגמה, עבור קרן גידור, יישמנו מודל ביצוע חלקי המבוסס על ספר הזמנות, מה ששיפר את דיוק התחזית ב-12%.

class RealisticBroker: def __init__( self, commission_pct: float = 0.001, # 0.1% slippage_pct: float = 0.0005, # 0.05% partial_fill_prob: float = 0.0, # 0 = всегда полное исполнение ): self.commission_pct = commission_pct self.slippage_pct = slippage_pct self.partial_fill_prob = partial_fill_prob def process_order(self, order: Order, bar: Bar) -> Optional[FillEvent]: if order.type == 'MARKET': base_price = bar.open slippage = base_price * self.slippage_pct fill_price = base_price + slippage if order.side == 'BUY' else base_price - slippage elif order.type == 'LIMIT': if order.side == 'BUY' and bar.low <= order.price: fill_price = min(order.price, bar.open) elif order.side == 'SELL' and bar.high >= order.price: fill_price = max(order.price, bar.open) else: return None elif order.type == 'STOP': if order.side == 'SELL' and bar.low <= order.stop_price: fill_price = min(order.stop_price, bar.open) fill_price -= fill_price * self.slippage_pct * 2 else: return None commission = fill_price * order.quantity * self.commission_pct return FillEvent( order_id=order.id, fill_price=fill_price, quantity=order.quantity, commission=commission, timestamp=bar.timestamp, ) 

איך עובדת לולאת הבדיקה הראשית?

עוברים על ברים: קודם מעבדים הזמנות ממתינות, ואז קוראים ל-class Portfolio: def __init__(self, initial_cash: float): self.initial_cash = initial_cash self.cash = initial_cash self.positions: dict[str, Position] = {} self.trades: list[dict] = [] self.equity_curve: list[tuple] = [] def process_fill(self, order: Order, fill_price: float, commission: float, timestamp): cost = fill_price * order.quantity if order.side == 'BUY': self.cash -= (cost + commission) symbol = order.symbol if symbol in self.positions: pos = self.positions[symbol] total_qty = pos.quantity + order.quantity pos.avg_entry_price = ( pos.avg_entry_price * pos.quantity + fill_price * order.quantity ) / total_qty pos.quantity = total_qty else: self.positions[symbol] = Position( symbol=symbol, side='LONG', quantity=order.quantity, avg_entry_price=fill_price, ) elif order.side == 'SELL': self.cash += (cost - commission) pos = self.positions.get(order.symbol) if pos: realized_pnl = (fill_price - pos.avg_entry_price) * order.quantity - commission pos.quantity -= order.quantity pos.realized_pnl += realized_pnl self.trades.append({ 'timestamp': timestamp, 'symbol': order.symbol, 'entry': pos.avg_entry_price, 'exit': fill_price, 'quantity': order.quantity, 'pnl': realized_pnl, }) if pos.quantity <= 0: del self.positions[order.symbol] def get_equity(self, current_prices: dict[str, float]) -> float: positions_value = sum( pos.quantity * current_prices.get(symbol, pos.avg_entry_price) for symbol, pos in self.positions.items() ) return self.cash + positions_value . הזמנות חדשות מהאסטרטגיה מתווספות לתור. בסוף כל בר, אנו רושמים הון.

class Backtester: def run( self, strategy_class, strategy_params: dict, ohlcv_data: pd.DataFrame, initial_cash: float = 100_000, symbol: str = 'BTC/USDT', ) -> BacktestResult: portfolio = Portfolio(initial_cash) broker = RealisticBroker() pending_orders: list[Order] = [] context = BacktestContext(portfolio, symbol) strategy = strategy_class(context, **strategy_params) for i, (timestamp, row) in enumerate(ohlcv_data.iterrows()): bar = Bar(timestamp=timestamp, **row.to_dict()) still_pending = [] for order in pending_orders: fill = broker.process_order(order, bar) if fill: portfolio.process_fill(order, fill.fill_price, fill.commission, timestamp) else: still_pending.append(order) pending_orders = still_pending for pos in portfolio.positions.values(): pos.unrealized_pnl = (bar.close - pos.avg_entry_price) * pos.quantity context.current_bar = bar strategy.on_bar(bar) pending_orders.extend(context.pop_new_orders()) equity = portfolio.get_equity({symbol: bar.close}) portfolio.equity_curve.append((timestamp, equity)) equity_series = pd.Series( [e for _, e in portfolio.equity_curve], index=[t for t, _ in portfolio.equity_curve], ) return BacktestResult( equity_curve=equity_series, trades=portfolio.trades, metrics=calculate_metrics(equity_series, portfolio.trades), ) 

ביצועים

למעבר על אלפי שילובי פרמטרים, אנו מייעלים צווארי בקבוק:

• וקטוריזציה עם NumPy עבור אינדיקטורים (SMA, RSI)—החלפת לולאות Python במערכים.
• JIT של Numba לחישובים קריטיים.
• ריבוי תהליכים—ריצות מקבילות על כל הליבות.
• טעינת נתונים במנות—טעינת נתונים בחלקים, לא הכל בזיכרון.

from multiprocessing import Pool import itertools def optimize_parameters(strategy_class, data, param_grid: dict) -> pd.DataFrame: combinations = list(itertools.product(*param_grid.values())) param_names = list(param_grid.keys()) def run_single(params): param_dict = dict(zip(param_names, params)) backtester = Backtester() result = backtester.run(strategy_class, param_dict, data) return {**param_dict, **result.metrics.__dict__} with Pool(processes=8) as pool: results = pool.map(run_single, combinations) return pd.DataFrame(results).sort_values('sharpe_ratio', ascending=False) 

על 8 ליבות, מעבר על 1000 שילובי פרמטרים עם סט נתונים שנתי לוקח 10–30 דקות תלוי במורכבות האסטרטגיה.

אילו מדדים מחושבים?

לאחר כל ריצה, אנו מחשבים סט סטנדרטי: תשואה כוללת, יחס שארפ, יחס סורטינו, משיכה מקסימלית, אחוז עסקאות רווחיות, מקדם התאוששות. ניתן להוסיף מדדים מותאמים אישית לפי בקשה—לדוגמה, יחס קלמר או מתאם נע עם מדד ייחוס. כל המדדים נשמרים במבנה RealisticBroker וניתן לייצא אותם ל-CSV לניתוח נוסף. הזמינו פיתוח וקבלו מדדים אמינים תוך 5–30 ימים.

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

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