자동매매 프로그램 시그널 및 워커 아키텍쳐

안녕하세요. 자동매매를 연구하며 개발하고 있는 인천고래입니다.

과거의 단순 코드 구조를 벗어나서 (기존 방식은 생성 순서와 의존성 문제 존재.)

키움 API를 통해서 전달되는 실시간 주가를 리얼타임으로 분석하여

다중 전략을 적용하기 쉬운 방향으로 아키텍쳐를 고려중이며 이에 현재 반영하고 있는 구조에 대해 공유하고자 글을 적습니다.

1. 기본 구조

레이어	하는 일	예시 위치
strategy	- 매수/매도 Logic 만 구현 - I/O · DB · 큐와 무관	module/strategy/ma_cross_20_60.py, …/price_cross_ma.py
worker (runner)	- DB 폴링·실시간 큐 구독 등 “데이터 수집 → strategy 호출 → router.publish” - 전략 클래스를 구성품으로 추가	module/worker/ma20x60_worker.py, …/ma_breakout_worker.py
infra / db	- DB 접속, 모델, 유틸	module/db/real_stock_data.db, SQL 헬퍼 등

장점
- 전략을 백테스트·시뮬레이션·실거래 등 다양한 환경에 재사용할 수 있음
- 워커는 데이터 소스나 메시지 중계 방식이 달라져도 전략 코드가 전혀 바뀌지 않음
- 단위 테스트가 용이 (전략만 테스트할 때 DB나 스레드를 띄울 필요가 없음)

컴포넌트“무엇을”“왜 필요한가” (책임)

Queue (queue.Queue)	스레드 간 안전한 버퍼 예) price_q, broadcast_q	생산자-소비자 사이의 완충(Back-pressure)·동기화
SignalWorker	price_q 에서 Tick 꺼내 전략 객체들에 전달	다중 전략을 한 스레드에서 실행해 시그널 딕셔너리를 생성
Strategy (BaseStrategy 상속)	“주가가 MA 돌파했나?” 등 판단 로직	로직만 갖고 외부 시스템과 분리 → 유닛테스트·플러그인용
Router (BroadcastRouter)	시그널을 다음 파이프라인으로 내보내는 인터페이스	로그 남기고, JSON 직렬화 후 broadcast_q에 put
PriceBroadcastWorker	price_q에서 Tick → {kind:"price", …} JSON 변환	가격도 시그널과 같은 큐로 보내 프런트가 한 스트림만 구독
SSE 라우트 (/SSE_Price/)	broadcast_q 소비 → HTTP Server-Sent Events	JS EventSource가 실시간 가격+시그널 모두 수신

다이어그램 읽는 법

RealTime DB → price_q Queue
키움AP를 통해 실시간 Tick 데이터가 들어오면 데이터베이스에 저장됨과 동시에 put(tick) 으로 파이프에 적재됩니다.
SignalWorker
price_q.get()으로 Tick 을 받아 여러 Strategy(MA 돌파, NetFlow 등)를 실행 → 매수·매도 시그널 생성 → BroadcastRouter.publish(signal).
PriceBroadcastWorker
Tick 중에서 UI에 필요-최소 가격만 골라 {kind:"price"} JSON 으로 만들어 broadcast_q 로 전송.
BroadcastRouter & broadcast_q
가격(kind:"price")과 시그널(kind:"signal")이 한 큐에 합류 → SSE 라우트가 소비.
Flask SSE /SSE_Price/ → Frontend JS EventSource
JS 코드는처럼 분기해 실시간 테이블·차트·알림을 갱신합니다.

if(msg.kind==="price") updatePrice(msg);
else if(msg.kind==="signal") showSignal(msg);

핵심 메모

Queue = 스레드 간 안전한 파이프

Worker = 데이터 소비·가공 엔진

Router = 시그널을 다음 파이프(큐/DB/Kafka)로 중계

이 구조를 통해 새 전략, 다른 데이터 소스, 추가 출력 경로를 큐·워커·라우터 조합으로 쉽게 확장할 수 있습니다.

감사합니다.