Coverage for src/finbot/engine/paper.py: 100%

1import hashlib 

2from dataclasses import replace 

3from datetime import datetime 

4 

5from finbot.engine.orchestrator import NoAction, TradingOrchestrator, TradingRejection 

6from finbot.execution.broker import PortfolioBroker 

7from finbot.execution.models import Order 

8from finbot.execution.runtime import PaperTradingRuntime 

9from finbot.market.data_provider import MarketDataProvider 

10from finbot.market.models import Bar 

11 

12 

13class PaperTradingCoordinator: 

14 """connects market data and trading orchestrator to paper execution""" 

15 

16 def __init__( 

17 self, 

18 market_data: MarketDataProvider, 

19 portfolio_broker: PortfolioBroker, 

20 orchestrator: TradingOrchestrator, 

21 runtime: PaperTradingRuntime, 

22 symbols: list[str], 

23 ): 

24 if not symbols: 

25 raise ValueError("At least one symbol is required") 

26 

27 self.market_data = market_data 

28 self.portfolio_broker = portfolio_broker 

29 self.orchestrator = orchestrator 

30 self.runtime = runtime 

31 self.symbols = symbols 

32 

33 self._last_processed: dict[str, datetime] = {} 

34 

35 def run_once(self) -> list[Order | TradingRejection | NoAction]: 

36 """Process the latest available bar for each symbol once""" 

37 

38 bars = self.market_data.get_latest_bars(self.symbols) 

39 market_prices = {symbol: bar.close for symbol, bar in bars.items()} 

40 

41 results: list[Order | TradingRejection | NoAction] = [] 

42 

43 for symbol in self.symbols: 

44 bar = bars[symbol] 

45 

46 last_processed = self._last_processed.get(symbol) 

47 if last_processed is not None and bar.timestamp <= last_processed: 

48 continue 

49 

50 signals = self.orchestrator.generate_signals(bar) 

51 

52 # The strategy has now "consumed" this bar; immediately mark it such that 

53 # a later execution/risk failure can't cause stateful strategies to 

54 # consume the same market event twice 

55 self._last_processed[symbol] = bar.timestamp 

56 

57 for signal_index, signal in enumerate(signals): 

58 portfolio = self.portfolio_broker.get_portfolio_snapshot() 

59 result = self.orchestrator.process_signal(signal, portfolio, market_prices) 

60 

61 if not isinstance(result, Order): 

62 results.append(result) 

63 continue 

64 

65 order = replace(result, client_order_id=self._client_order_id(bar, signal_index)) 

66 

67 self.runtime.submit_order(order) 

68 results.append(order) 

69 

70 return results 

71 

72 @staticmethod 

73 def _client_order_id(bar: Bar, signal_index: int) -> str: 

74 event_key = f"{bar.symbol}|{bar.timestamp.isoformat()}|{signal_index}" 

75 digest = hashlib.sha256(event_key.encode("utf-8")).hexdigest()[:24] 

76 return f"finbot-{digest}"