Space-Based Architecture

1
2

Space = {(key, value, metadata) | 모든 튜플}
Operations = {out, in, rd, eval}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

# JavaSpaces 기반 튜플 스페이스 구현 예시
class TupleSpace:
    def __init__(self):
        self.tuples = []
        self.lock = threading.Lock()
    
    def out(self, tuple_data):
        """튜플을 스페이스에 추가"""
        with self.lock:
            self.tuples.append(tuple_data)
    
    def in_tuple(self, template):
        """패턴에 맞는 튜플을 제거하며 반환"""
        with self.lock:
            for i, tuple_data in enumerate(self.tuples):
                if self.match(tuple_data, template):
                    return self.tuples.pop(i)
        return None
    
    def rd(self, template):
        """패턴에 맞는 튜플을 복사하여 반환"""
        with self.lock:
            for tuple_data in self.tuples:
                if self.match(tuple_data, template):
                    return tuple_data
        return None

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

class DataPartitioner:
    def __init__(self, num_partitions):
        self.num_partitions = num_partitions
        self.partitions = [[] for _ in range(num_partitions)]
    
    def partition_key(self, key):
        """키를 기반으로 파티션 결정"""
        return hash(key) % self.num_partitions
    
    def put(self, key, value):
        """데이터를 적절한 파티션에 저장"""
        partition_id = self.partition_key(key)
        self.partitions[partition_id].append((key, value))
    
    def get(self, key):
        """키를 기반으로 데이터 조회"""
        partition_id = self.partition_key(key)
        for k, v in self.partitions[partition_id]:
            if k == key:
                return v
        return None

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

class EventDrivenProcessor:
    def __init__(self):
        self.event_handlers = {}
        self.event_queue = queue.Queue()
    
    def register_handler(self, event_type, handler):
        """이벤트 핸들러 등록"""
        if event_type not in self.event_handlers:
            self.event_handlers[event_type] = []
        self.event_handlers[event_type].append(handler)
    
    def publish_event(self, event_type, data):
        """이벤트 발행"""
        self.event_queue.put((event_type, data))
    
    def process_events(self):
        """이벤트 처리"""
        while not self.event_queue.empty():
            event_type, data = self.event_queue.get()
            if event_type in self.event_handlers:
                for handler in self.event_handlers[event_type]:
                    handler(data)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

class DistributedCache:
    def __init__(self, nodes):
        self.nodes = nodes
        self.ring = ConsistentHash(nodes)
    
    def put(self, key, value):
        """데이터를 분산 캐시에 저장"""
        node = self.ring.get_node(key)
        node.put(key, value)
        # 복제를 위해 추가 노드에도 저장
        replica_node = self.ring.get_replica_node(key)
        replica_node.put(key, value)
    
    def get(self, key):
        """분산 캐시에서 데이터 조회"""
        node = self.ring.get_node(key)
        value = node.get(key)
        if value is None:
            # 복제 노드에서 조회 시도
            replica_node = self.ring.get_replica_node(key)
            value = replica_node.get(key)
        return value

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# 주문 ID 기반으로 공간(Space)을 결정하는 예시
class SpaceRouter:
    def __init__(self, space_count):
        self.space_count = space_count

    def route(self, order_id):
        # order_id를 해시하여 공간 번호 결정
        return order_id % self.space_count

# 사용 예시
space_router = SpaceRouter(space_count=2)
order_id = 20230711
space_number = space_router.route(order_id)
print(f"주문 {order_id}는 공간 {space_number}로 라우팅됩니다.")

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286

import time
import threading
from collections import defaultdict
from queue import Queue
from typing import Dict, List, Optional
import json

class TradingData:
    """거래 데이터 클래스"""
    def __init__(self, symbol: str, price: float, volume: int, timestamp: float):
        self.symbol = symbol
        self.price = price
        self.volume = volume
        self.timestamp = timestamp

class Position:
    """포지션 데이터 클래스"""
    def __init__(self, symbol: str, quantity: int, avg_price: float):
        self.symbol = symbol
        self.quantity = quantity
        self.avg_price = avg_price
        self.pnl = 0.0

class InMemoryDataGrid:
    """Space-Based Architecture의 인메모리 데이터 그리드"""
    def __init__(self):
        self.data = {}  # 메모리 내 데이터 저장소
        self.lock = threading.RLock()  # 동시성 제어
        self.event_handlers = defaultdict(list)  # 이벤트 핸들러
    
    def put(self, key: str, value: any) -> None:
        """데이터 저장 및 이벤트 발생"""
        with self.lock:
            old_value = self.data.get(key)
            self.data[key] = value
            # 데이터 변경 이벤트 발생
            self._fire_event('data_changed', key, old_value, value)
    
    def get(self, key: str) -> Optional[any]:
        """데이터 조회"""
        with self.lock:
            return self.data.get(key)
    
    def register_event_handler(self, event_type: str, handler) -> None:
        """이벤트 핸들러 등록"""
        self.event_handlers[event_type].append(handler)
    
    def _fire_event(self, event_type: str, *args) -> None:
        """이벤트 발생"""
        for handler in self.event_handlers[event_type]:
            try:
                handler(*args)
            except Exception as e:
                print(f"Event handler error: {e}")

class TradingProcessingUnit:
    """거래 처리 단위 (Processing Unit)"""
    def __init__(self, symbol: str, data_grid: InMemoryDataGrid):
        self.symbol = symbol
        self.data_grid = data_grid
        self.positions = {}  # 포지션 관리
        self.market_data_queue = Queue()  # 시장 데이터 큐
        self.is_running = False
        self.processing_thread = None
        
        # 시장 데이터 변경 이벤트 핸들러 등록
        self.data_grid.register_event_handler(
            'data_changed', 
            self._on_market_data_changed
        )
    
    def start(self) -> None:
        """처리 단위 시작"""
        if not self.is_running:
            self.is_running = True
            self.processing_thread = threading.Thread(target=self._process_market_data)
            self.processing_thread.start()
            print(f"Trading PU for {self.symbol} started")
    
    def stop(self) -> None:
        """처리 단위 중지"""
        self.is_running = False
        if self.processing_thread:
            self.processing_thread.join()
        print(f"Trading PU for {self.symbol} stopped")
    
    def _on_market_data_changed(self, key: str, old_value: any, new_value: any) -> None:
        """시장 데이터 변경 이벤트 처리"""
        if key.startswith(f"market_data_{self.symbol}"):
            self.market_data_queue.put(new_value)
    
    def _process_market_data(self) -> None:
        """시장 데이터 처리 스레드"""
        while self.is_running:
            try:
                # 시장 데이터 대기 (타임아웃 설정)
                market_data = self.market_data_queue.get(timeout=0.1)
                
                # 거래 알고리즘 실행
                trading_signal = self._generate_trading_signal(market_data)
                
                # 거래 실행
                if trading_signal:
                    self._execute_trade(trading_signal)
                
                # 포지션 업데이트
                self._update_positions()
                
            except Exception as e:
                if self.is_running:  # 정상 종료가 아닌 경우만 에러 출력
                    print(f"Processing error in {self.symbol}: {e}")
    
    def _generate_trading_signal(self, market_data: TradingData) -> Optional[Dict]:
        """거래 신호 생성 (간단한 알고리즘 예시)"""
        # 현재 포지션 조회
        current_position = self.positions.get(self.symbol, Position(self.symbol, 0, 0.0))
        
        # 간단한 평균 회귀 전략
        historical_data = self.data_grid.get(f"historical_{self.symbol}")
        if historical_data and len(historical_data) > 10:
            avg_price = sum(data.price for data in historical_data[-10:]) / 10
            
            # 현재 가격이 평균보다 5% 이상 낮으면 매수
            if market_data.price < avg_price * 0.95 and current_position.quantity < 1000:
                return {
                    'action': 'BUY',
                    'symbol': self.symbol,
                    'quantity': 100,
                    'price': market_data.price
                }
            
            # 현재 가격이 평균보다 5% 이상 높으면 매도
            elif market_data.price > avg_price * 1.05 and current_position.quantity > 0:
                return {
                    'action': 'SELL',
                    'symbol': self.symbol,
                    'quantity': min(100, current_position.quantity),
                    'price': market_data.price
                }
        
        return None
    
    def _execute_trade(self, signal: Dict) -> None:
        """거래 실행"""
        print(f"Executing trade: {signal}")
        
        # 포지션 업데이트
        position = self.positions.get(signal['symbol'], Position(signal['symbol'], 0, 0.0))
        
        if signal['action'] == 'BUY':
            new_quantity = position.quantity + signal['quantity']
            new_avg_price = ((position.avg_price * position.quantity) + 
                           (signal['price'] * signal['quantity'])) / new_quantity
            position.quantity = new_quantity
            position.avg_price = new_avg_price
        
        elif signal['action'] == 'SELL':
            position.quantity -= signal['quantity']
            if position.quantity < 0:
                position.quantity = 0
        
        self.positions[signal['symbol']] = position
        
        # 데이터 그리드에 포지션 업데이트
        self.data_grid.put(f"position_{self.symbol}", position)
    
    def _update_positions(self) -> None:
        """포지션 업데이트"""
        for symbol, position in self.positions.items():
            current_price = self.data_grid.get(f"market_data_{symbol}")
            if current_price:
                position.pnl = (current_price.price - position.avg_price) * position.quantity
                self.data_grid.put(f"position_{symbol}", position)

class MessageGrid:
    """메시지 그리드 - 요청 라우팅 담당"""
    def __init__(self):
        self.processing_units = {}  # 심볼별 처리 단위 매핑
        self.data_grid = InMemoryDataGrid()
    
    def register_processing_unit(self, symbol: str, processing_unit: TradingProcessingUnit) -> None:
        """처리 단위 등록"""
        self.processing_units[symbol] = processing_unit
    
    def route_market_data(self, market_data: TradingData) -> None:
        """시장 데이터 라우팅"""
        # 데이터 그리드에 시장 데이터 저장
        self.data_grid.put(f"market_data_{market_data.symbol}", market_data)
        
        # 이력 데이터 업데이트
        historical_key = f"historical_{market_data.symbol}"
        historical_data = self.data_grid.get(historical_key) or []
        historical_data.append(market_data)
        
        # 최근 100개 데이터만 유지
        if len(historical_data) > 100:
            historical_data = historical_data[-100:]
        
        self.data_grid.put(historical_key, historical_data)

class SpaceBasedTradingSystem:
    """Space-Based Architecture 기반 거래 시스템"""
    def __init__(self):
        self.message_grid = MessageGrid()
        self.processing_units = {}
        self.is_running = False
    
    def add_trading_symbol(self, symbol: str) -> None:
        """거래 심볼 추가"""
        pu = TradingProcessingUnit(symbol, self.message_grid.data_grid)
        self.processing_units[symbol] = pu
        self.message_grid.register_processing_unit(symbol, pu)
    
    def start(self) -> None:
        """시스템 시작"""
        self.is_running = True
        for pu in self.processing_units.values():
            pu.start()
        print("Space-Based Trading System started")
    
    def stop(self) -> None:
        """시스템 중지"""
        self.is_running = False
        for pu in self.processing_units.values():
            pu.stop()
        print("Space-Based Trading System stopped")
    
    def feed_market_data(self, symbol: str, price: float, volume: int) -> None:
        """시장 데이터 피드"""
        if self.is_running:
            market_data = TradingData(symbol, price, volume, time.time())
            self.message_grid.route_market_data(market_data)
    
    def get_positions(self) -> Dict[str, Position]:
        """모든 포지션 조회"""
        positions = {}
        for symbol in self.processing_units.keys():
            position = self.message_grid.data_grid.get(f"position_{symbol}")
            if position:
                positions[symbol] = position
        return positions

# 사용 예시
if __name__ == "__main__":
    # Space-Based Trading System 초기화
    trading_system = SpaceBasedTradingSystem()
    
    # 거래 심볼 추가
    trading_system.add_trading_symbol("AAPL")
    trading_system.add_trading_symbol("GOOGL")
    trading_system.add_trading_symbol("MSFT")
    
    # 시스템 시작
    trading_system.start()
    
    # 시장 데이터 시뮬레이션
    import random
    
    try:
        symbols = ["AAPL", "GOOGL", "MSFT"]
        base_prices = {"AAPL": 150.0, "GOOGL": 2500.0, "MSFT": 300.0}
        
        for i in range(1000):
            symbol = random.choice(symbols)
            price_change = random.uniform(-0.05, 0.05)  # ±5% 변동
            new_price = base_prices[symbol] * (1 + price_change)
            volume = random.randint(100, 1000)
            
            trading_system.feed_market_data(symbol, new_price, volume)
            base_prices[symbol] = new_price
            
            # 포지션 상태 주기적 출력
            if i % 100 == 0:
                positions = trading_system.get_positions()
                print(f"\n=== Positions at iteration {i} ===")
                for symbol, position in positions.items():
                    print(f"{symbol}: Qty={position.quantity}, "
                          f"Avg=${position.avg_price:f}, PnL=${position.pnl:f}")
            
            time.sleep(0.01)  # 10ms 간격
    
    except KeyboardInterrupt:
        print("\nShutting down…")
    
    finally:
        trading_system.stop()