25 KiB
25 KiB
Design: ICT Crypto Trading Bot
Feature: ict-crypto-bot Created: 2026-03-17 Phase: Design Plan Reference:
docs/01-plan/features/ict-crypto-bot.plan.md
1. Project Structure
crypto_news/
├── config/
│ ├── settings.py # 전역 설정 (API keys, 기본값)
│ ├── trading_pairs.py # 거래쌍 설정
│ └── strategies.py # 전략 파라미터
├── core/
│ ├── __init__.py
│ ├── bot.py # 메인 봇 오케스트레이터
│ ├── data_feed.py # 데이터 수집 (CCXT/WebSocket)
│ └── event_bus.py # 이벤트 기반 통신
├── indicators/
│ ├── __init__.py
│ ├── ict_engine.py # ICT 지표 통합 엔진
│ ├── multi_timeframe.py # 멀티 타임프레임 분석
│ └── confluence.py # 신호 합류 판단
├── strategy/
│ ├── __init__.py
│ ├── signal_generator.py # 매매 신호 생성
│ ├── entry_rules.py # 진입 규칙
│ └── exit_rules.py # 청산 규칙
├── execution/
│ ├── __init__.py
│ ├── order_manager.py # 주문 실행/관리
│ ├── position_manager.py # 포지션 관리
│ └── exchange_client.py # 거래소 클라이언트 래퍼
├── risk/
│ ├── __init__.py
│ ├── risk_manager.py # 리스크 관리 엔진
│ ├── position_sizing.py # 포지션 사이징
│ └── drawdown_monitor.py # 낙폭 모니터링
├── backtest/
│ ├── __init__.py
│ ├── backtester.py # 백테스트 엔진
│ ├── data_loader.py # 과거 데이터 로더
│ └── performance.py # 성과 분석
├── notification/
│ ├── __init__.py
│ ├── telegram_bot.py # 텔레그램 알림
│ └── alert_manager.py # 알림 관리
├── dashboard/
│ └── app.py # Streamlit 대시보드
├── database/
│ ├── __init__.py
│ ├── models.py # DB 모델
│ └── repository.py # 데이터 접근 레이어
├── tests/
│ ├── test_ict_engine.py
│ ├── test_signal_generator.py
│ ├── test_order_manager.py
│ └── test_risk_manager.py
├── main.py # 엔트리 포인트
├── requirements.txt
├── .env.example
└── README.md
2. Module Design
2.1 Data Feed Module (core/data_feed.py)
class DataFeed:
"""실시간 + 히스토리 데이터 수집"""
async def connect(exchange_id: str, api_key: str, secret: str) -> None
async def disconnect() -> None
# 실시간 데이터 (WebSocket)
async def watch_ohlcv(symbol: str, timeframe: str) -> List[OHLCV]
async def watch_ticker(symbol: str) -> Ticker
async def watch_order_book(symbol: str) -> OrderBook
# 히스토리 데이터 (REST)
async def fetch_ohlcv(symbol: str, timeframe: str, since: int, limit: int) -> pd.DataFrame
# 멀티 타임프레임 데이터 유지
def get_dataframe(symbol: str, timeframe: str) -> pd.DataFrame
데이터 흐름:
Exchange WebSocket
│
▼
DataFeed.watch_ohlcv()
│
▼ (pd.DataFrame: open, high, low, close, volume)
ICTEngine.analyze()
│
▼ (ICTSignals 객체)
SignalGenerator.evaluate()
│
▼ (TradeSignal: BUY/SELL/HOLD)
OrderManager.execute()
2.2 ICT Indicator Engine (indicators/ict_engine.py)
from smartmoneyconcepts import smc
class ICTEngine:
"""ICT Smart Money Concepts 지표 통합 엔진"""
def __init__(self, swing_length: int = 50):
self.swing_length = swing_length
def analyze(self, ohlc: pd.DataFrame) -> ICTSignals:
"""전체 ICT 분석 실행, ICTSignals 객체 반환"""
swing = smc.swing_highs_lows(ohlc, self.swing_length)
return ICTSignals(
swing_highs_lows = swing,
fvg = smc.fvg(ohlc, join_consecutive=False),
bos_choch = smc.bos_choch(ohlc, swing, close_break=True),
order_blocks = smc.ob(ohlc, swing, close_mitigation=False),
liquidity = smc.liquidity(ohlc, swing, range_percent=0.01),
prev_high_low = smc.previous_high_low(ohlc, time_frame="1D"),
retracements = smc.retracements(ohlc, swing),
)
@dataclass
class ICTSignals:
"""ICT 분석 결과 데이터 컨테이너"""
swing_highs_lows: pd.DataFrame # HighLow, Level
fvg: pd.DataFrame # FVG(1/-1), Top, Bottom, MitigatedIndex
bos_choch: pd.DataFrame # BOS(1/-1), CHoCH(1/-1), Level, BrokenIndex
order_blocks: pd.DataFrame # OB(1/-1), Top, Bottom, OBVolume, Percentage
liquidity: pd.DataFrame # Direction(1/-1), Level, End, SweptIndex
prev_high_low: pd.DataFrame # PreviousHigh, PreviousLow, BrokenHigh/Low
retracements: pd.DataFrame # Direction, CurrentRetracement, DeepestRetracement
2.3 Multi-Timeframe Analyzer (indicators/multi_timeframe.py)
class MultiTimeframeAnalyzer:
"""멀티 타임프레임 ICT 분석"""
TIMEFRAMES = {
'htf': '4h', # Higher Timeframe: 시장 방향 (bias)
'mtf': '1h', # Middle Timeframe: 구조 분석 + OB/FVG
'ltf': '15m', # Lower Timeframe: 정밀 진입
}
def analyze_all(self, data_feed: DataFeed, symbol: str) -> MTFAnalysis:
"""모든 타임프레임 분석 후 종합"""
def get_htf_bias(self, htf_signals: ICTSignals) -> MarketBias:
"""HTF에서 시장 방향 판단 (BULLISH / BEARISH / NEUTRAL)"""
def find_mtf_zones(self, mtf_signals: ICTSignals) -> List[TradeZone]:
"""MTF에서 OB, FVG 존 탐지"""
def find_ltf_entry(self, ltf_signals: ICTSignals, bias: MarketBias, zones: List[TradeZone]) -> Optional[EntryPoint]:
"""LTF에서 정밀 진입 타이밍 탐색"""
@dataclass
class MTFAnalysis:
htf_bias: MarketBias # BULLISH / BEARISH / NEUTRAL
mtf_zones: List[TradeZone] # OB/FVG 존 리스트
ltf_entry: Optional[EntryPoint] # 진입 포인트 (없으면 None)
confluence_score: int # 합류 점수 (0~6)
2.4 Confluence Checker (indicators/confluence.py)
class ConfluenceChecker:
"""ICT 신호 합류 판단 - 최소 3개 이상 조건 충족 시 유효"""
MIN_CONFLUENCE = 3 # 최소 합류 점수
def check(self, mtf: MTFAnalysis, current_price: float) -> ConfluenceResult:
"""
6가지 조건 체크:
1. Market Structure (HTF bias 일치)
2. Liquidity Sweep (스윕 후 반전)
3. Order Block (OB 존 진입)
4. Fair Value Gap (FVG 회귀)
5. BOS (구조 돌파 확인)
6. CHOCH (성격 변화 확인)
Returns: ConfluenceResult(score=0~6, conditions=[], is_valid=bool)
"""
@dataclass
class ConfluenceResult:
score: int # 0~6
conditions: List[ConditionResult] # 개별 조건 결과
is_valid: bool # score >= MIN_CONFLUENCE
direction: TradeDirection # LONG / SHORT / NONE
2.5 Signal Generator (strategy/signal_generator.py)
class SignalGenerator:
"""매매 신호 생성기"""
def __init__(self, ict_engine: ICTEngine, mtf_analyzer: MultiTimeframeAnalyzer,
confluence_checker: ConfluenceChecker):
...
async def generate(self, symbol: str, data_feed: DataFeed) -> TradeSignal:
"""
1. MTF 분석 실행
2. Confluence 체크
3. Entry/Exit 규칙 적용
4. TradeSignal 반환
"""
@dataclass
class TradeSignal:
symbol: str
direction: TradeDirection # LONG / SHORT
entry_price: float
stop_loss: float
take_profit: float
confidence: int # Confluence score (3~6)
timeframe: str
timestamp: datetime
reasons: List[str] # 진입 근거
2.6 Entry Rules (strategy/entry_rules.py)
class EntryRules:
"""ICT 진입 규칙"""
def check_bullish_entry(self, signals: ICTSignals, price: float) -> EntryResult:
"""
롱 진입 조건:
1. HTF: Higher Highs & Higher Lows
2. Liquidity Sweep: 이전 저점 스윕 후 반등
3. Order Block: Bullish OB 존에 가격 진입
4. FVG: Bullish FVG에서 가격 회귀
5. BOS: 상방 구조 돌파
"""
def check_bearish_entry(self, signals: ICTSignals, price: float) -> EntryResult:
"""숏 진입 조건 (반대 로직)"""
def calculate_stop_loss(self, direction: TradeDirection, signals: ICTSignals) -> float:
"""OB 너머 또는 최근 스윙 고/저점 기준 SL 계산"""
def calculate_take_profit(self, direction: TradeDirection, signals: ICTSignals, entry: float) -> float:
"""반대편 OB 또는 FVG 기준 TP 계산"""
2.7 Exit Rules (strategy/exit_rules.py)
class ExitRules:
"""ICT 청산 규칙"""
def should_exit(self, position: Position, signals: ICTSignals, price: float) -> ExitResult:
"""
청산 조건 체크:
1. TP 도달 (반대편 OB/FVG)
2. SL 도달
3. CHOCH 반대 방향 발생
4. Time-based exit (설정 시간 경과)
5. Trailing Stop 트리거
"""
def update_trailing_stop(self, position: Position, price: float) -> float:
"""수익 구간 진입 후 동적 SL 업데이트"""
2.8 Order Manager (execution/order_manager.py)
class OrderManager:
"""주문 실행 및 관리"""
def __init__(self, exchange_client: ExchangeClient, risk_manager: RiskManager):
...
async def execute_signal(self, signal: TradeSignal) -> Order:
"""
1. RiskManager 승인 확인
2. 포지션 사이징 계산
3. 주문 생성 (Limit 우선, 실패 시 Market)
4. SL/TP 주문 동시 설정
5. DB 기록
"""
async def create_order(self, symbol: str, side: str, order_type: str,
amount: float, price: float = None) -> Order:
"""거래소 주문 생성"""
async def cancel_order(self, order_id: str, symbol: str) -> bool
async def modify_order(self, order_id: str, price: float, amount: float) -> Order
2.9 Risk Manager (risk/risk_manager.py)
class RiskManager:
"""리스크 관리 엔진"""
# 설정값
MAX_RISK_PER_TRADE = 0.02 # 거래당 최대 2%
MAX_DAILY_LOSS = 0.05 # 일일 최대 손실 5%
MAX_CONCURRENT_POSITIONS = 3 # 동시 포지션 최대 3개
MAX_LEVERAGE = 3 # 최대 레버리지 3x
MAX_DRAWDOWN = 0.15 # 최대 낙폭 15%
def approve_trade(self, signal: TradeSignal, balance: float) -> RiskApproval:
"""
매매 승인 체크:
1. 일일 손실 한도 미초과
2. 동시 포지션 수 미초과
3. 최대 낙폭 미초과
4. 포지션 사이즈 적정성
"""
def calculate_position_size(self, balance: float, entry: float,
stop_loss: float, risk_pct: float) -> float:
"""
포지션 사이즈 = (Balance * Risk%) / |Entry - StopLoss|
"""
def update_daily_pnl(self, pnl: float) -> None
def check_drawdown(self, equity_curve: List[float]) -> bool
def emergency_stop(self) -> None:
"""모든 포지션 즉시 청산, 봇 정지"""
2.10 Exchange Client (execution/exchange_client.py)
class ExchangeClient:
"""CCXT 기반 거래소 클라이언트"""
def __init__(self, exchange_id: str = 'binance'):
self.exchange = ccxt.pro.binance({
'apiKey': settings.API_KEY,
'secret': settings.API_SECRET,
'sandbox': settings.SANDBOX_MODE, # 테스트넷
'options': {'defaultType': 'spot'},
})
# 데이터
async def watch_ohlcv(symbol, timeframe) -> List
async def fetch_ohlcv(symbol, timeframe, since, limit) -> pd.DataFrame
async def fetch_balance() -> dict
async def fetch_ticker(symbol) -> dict
# 주문
async def create_limit_buy(symbol, amount, price) -> dict
async def create_limit_sell(symbol, amount, price) -> dict
async def create_market_buy(symbol, amount) -> dict
async def create_market_sell(symbol, amount) -> dict
async def create_stop_loss(symbol, amount, stop_price) -> dict
async def cancel_order(order_id, symbol) -> dict
# 연결 관리
async def connect() -> None
async def disconnect() -> None
async def is_connected() -> bool
3. Data Models (database/models.py)
@dataclass
class Position:
id: str
symbol: str
direction: TradeDirection # LONG / SHORT
entry_price: float
current_price: float
amount: float
stop_loss: float
take_profit: float
trailing_stop: Optional[float]
unrealized_pnl: float
realized_pnl: float
status: PositionStatus # OPEN / CLOSED / LIQUIDATED
opened_at: datetime
closed_at: Optional[datetime]
close_reason: Optional[str] # TP / SL / CHOCH / TRAILING / TIME / MANUAL
confluence_score: int
entry_reasons: List[str]
@dataclass
class TradeRecord:
id: str
position_id: str
symbol: str
side: str # buy / sell
order_type: str # market / limit / stop
price: float
amount: float
fee: float
timestamp: datetime
@dataclass
class DailyPerformance:
date: str
total_trades: int
winning_trades: int
losing_trades: int
total_pnl: float
win_rate: float
max_drawdown: float
sharpe_ratio: float
DB Schema (SQLite)
CREATE TABLE positions (
id TEXT PRIMARY KEY,
symbol TEXT NOT NULL,
direction TEXT NOT NULL,
entry_price REAL NOT NULL,
amount REAL NOT NULL,
stop_loss REAL NOT NULL,
take_profit REAL NOT NULL,
trailing_stop REAL,
realized_pnl REAL DEFAULT 0,
status TEXT DEFAULT 'OPEN',
opened_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
closed_at TIMESTAMP,
close_reason TEXT,
confluence_score INTEGER,
entry_reasons TEXT
);
CREATE TABLE trade_records (
id TEXT PRIMARY KEY,
position_id TEXT REFERENCES positions(id),
symbol TEXT NOT NULL,
side TEXT NOT NULL,
order_type TEXT NOT NULL,
price REAL NOT NULL,
amount REAL NOT NULL,
fee REAL DEFAULT 0,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE daily_performance (
date TEXT PRIMARY KEY,
total_trades INTEGER DEFAULT 0,
winning_trades INTEGER DEFAULT 0,
losing_trades INTEGER DEFAULT 0,
total_pnl REAL DEFAULT 0,
max_drawdown REAL DEFAULT 0
);
CREATE TABLE bot_state (
key TEXT PRIMARY KEY,
value TEXT,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
4. Core Flow Diagram
4.1 Main Bot Loop
┌─────────────────────────────────────────────────────────────────┐
│ Bot.run() - Main Loop │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌───────────┐ ┌──────────────┐ │
│ │ DataFeed │───▶│ ICTEngine │───▶│ MTFAnalyzer │ │
│ │ watch_ │ │ analyze() │ │ analyze_all()│ │
│ │ ohlcv() │ └───────────┘ └──────┬───────┘ │
│ └──────────┘ │ │
│ ▼ │
│ ┌───────────────────┐ │
│ │ ConfluenceChecker │ │
│ │ check() >= 3? │ │
│ └────────┬──────────┘ │
│ │ │
│ ┌────────────┼────────────┐ │
│ │ YES │ │ NO │
│ ▼ │ ▼ │
│ ┌──────────────┐ │ ┌────────────┐ │
│ │SignalGenerator│ │ │ Continue │ │
│ │ generate() │ │ │ Watching │ │
│ └──────┬───────┘ │ └────────────┘ │
│ │ │ │
│ ▼ │ │
│ ┌──────────────┐ │ │
│ │ RiskManager │ │ │
│ │ approve? │ │ │
│ └──────┬───────┘ │ │
│ │ │ │
│ YES ▼ NO ▼ │
│ ┌──────────────┐ ┌────────┐ │
│ │OrderManager │ │ Skip │ │
│ │ execute() │ │ + Log │ │
│ └──────┬───────┘ └────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │PositionMgr │ │
│ │ + AlertMgr │ │
│ │ + DB Record │ │
│ └──────────────┘ │
└─────────────────────────────────────────────────────────────────┘
4.2 Position Lifecycle
Signal Generated
│
▼
RiskManager.approve_trade()
│
├── REJECTED → Log + Skip
│
▼ APPROVED
OrderManager.execute_signal()
│
▼
Position OPEN
│
├── ExitRules.should_exit() [매 캔들마다 체크]
│ ├── TP Hit → Close + Record
│ ├── SL Hit → Close + Record
│ ├── CHOCH → Close + Record
│ ├── Trailing Stop → Close + Record
│ └── Time Exit → Close + Record
│
▼
Position CLOSED
│
▼
DailyPerformance Update
│
▼
Telegram Notification
5. Configuration Design (config/settings.py)
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
# Exchange
EXCHANGE_ID: str = "binance"
API_KEY: str
API_SECRET: str
SANDBOX_MODE: bool = True # True = 테스트넷
# Trading
TRADING_PAIRS: list = ["BTC/USDT", "ETH/USDT"]
DEFAULT_LEVERAGE: int = 1
MAX_LEVERAGE: int = 3
# ICT Parameters
SWING_LENGTH: int = 50
FVG_JOIN_CONSECUTIVE: bool = False
OB_CLOSE_MITIGATION: bool = False
LIQUIDITY_RANGE_PERCENT: float = 0.01
MIN_CONFLUENCE_SCORE: int = 3
# Timeframes
HTF_TIMEFRAME: str = "4h"
MTF_TIMEFRAME: str = "1h"
LTF_TIMEFRAME: str = "15m"
# Risk Management
MAX_RISK_PER_TRADE: float = 0.02 # 2%
MAX_DAILY_LOSS: float = 0.05 # 5%
MAX_CONCURRENT_POSITIONS: int = 3
MAX_DRAWDOWN: float = 0.15 # 15%
# Notification
TELEGRAM_BOT_TOKEN: str = ""
TELEGRAM_CHAT_ID: str = ""
# Database
DB_PATH: str = "data/trading.db"
# Logging
LOG_LEVEL: str = "INFO"
LOG_FILE: str = "logs/bot.log"
class Config:
env_file = ".env"
6. Backtest Design (backtest/backtester.py)
class Backtester:
"""ICT 전략 백테스트 엔진"""
def __init__(self, strategy: SignalGenerator, risk_manager: RiskManager):
...
def run(self, symbol: str, timeframe: str,
start_date: str, end_date: str,
initial_balance: float = 1000) -> BacktestResult:
"""
1. 과거 데이터 로드
2. 캔들별 순회하며 전략 실행
3. 가상 주문 체결
4. 성과 계산
"""
def generate_report(self, result: BacktestResult) -> dict:
"""
반환 지표:
- Total PnL, ROI %
- Win Rate
- Profit Factor
- Sharpe Ratio
- Max Drawdown
- Average Trade Duration
- Total Trades / Win / Loss
"""
@dataclass
class BacktestResult:
trades: List[BacktestTrade]
equity_curve: List[float]
total_pnl: float
roi_percent: float
win_rate: float
profit_factor: float
sharpe_ratio: float
max_drawdown: float
total_trades: int
avg_trade_duration: timedelta
7. Notification Design (notification/telegram_bot.py)
class TelegramNotifier:
"""텔레그램 매매 알림"""
async def send_signal(self, signal: TradeSignal) -> None:
"""
📊 ICT Signal Detected
──────────────────
Symbol: BTC/USDT
Direction: LONG
Entry: $67,500
SL: $66,800 (-1.04%)
TP: $69,200 (+2.52%)
Confluence: 4/6
Reasons: OB + FVG + BOS + Liquidity Sweep
"""
async def send_fill(self, order: Order) -> None
async def send_close(self, position: Position) -> None
async def send_daily_report(self, performance: DailyPerformance) -> None
async def send_error(self, error: str) -> None
async def send_emergency(self, msg: str) -> None
8. Implementation Order
| 순서 | 모듈 | 파일 | 의존성 |
|---|---|---|---|
| 1 | 프로젝트 초기화 | requirements.txt, .env, config/ | 없음 |
| 2 | 거래소 클라이언트 | execution/exchange_client.py | config |
| 3 | 데이터 수집 | core/data_feed.py | exchange_client |
| 4 | ICT 지표 엔진 | indicators/ict_engine.py | smartmoneyconcepts |
| 5 | 멀티 타임프레임 | indicators/multi_timeframe.py | ict_engine, data_feed |
| 6 | 합류 체커 | indicators/confluence.py | ict_engine |
| 7 | 신호 생성기 | strategy/signal_generator.py | mtf, confluence |
| 8 | 진입/청산 규칙 | strategy/entry_rules.py, exit_rules.py | ict_engine |
| 9 | 리스크 관리 | risk/risk_manager.py | config |
| 10 | 주문 관리 | execution/order_manager.py | exchange_client, risk |
| 11 | 포지션 관리 | execution/position_manager.py | order_manager |
| 12 | DB 모델 | database/models.py, repository.py | 없음 |
| 13 | 백테스트 | backtest/backtester.py | strategy, risk |
| 14 | 알림 | notification/telegram_bot.py | 없음 |
| 15 | 대시보드 | dashboard/app.py | database |
| 16 | 메인 봇 | core/bot.py, main.py | 모든 모듈 |
9. Dependencies (requirements.txt)
# Core
ccxt>=4.0.0
pandas>=2.0.0
numpy>=1.24.0
# ICT Indicators
smartmoneyconcepts>=0.1.0
# Async
asyncio
aiohttp>=3.9.0
# Configuration
pydantic-settings>=2.0.0
python-dotenv>=1.0.0
# Database
aiosqlite>=0.19.0
# Notification
python-telegram-bot>=20.0
# Dashboard
streamlit>=1.30.0
plotly>=5.18.0
# Backtest
matplotlib>=3.8.0
# Logging
loguru>=0.7.0
# Scheduling
apscheduler>=3.10.0
10. Error Handling Strategy
| 에러 유형 | 처리 방식 |
|---|---|
| WebSocket 연결 끊김 | 자동 재연결 (CCXT 내장) + 로그 |
| 주문 실패 | 3회 재시도 → 실패 시 Telegram 알림 |
| API Rate Limit | CCXT 내장 레이트 리밋 + 백오프 |
| 잔고 부족 | 주문 스킵 + 알림 |
| 거래소 점검 | 봇 일시 중지 + 알림 |
| 예상치 못한 에러 | Emergency Stop + 전체 포지션 청산 + 알림 |
11. Security Considerations
| 항목 | 방법 |
|---|---|
| API Key 관리 | .env 파일 (git 제외), 환경변수 |
| API 권한 | 거래 권한만 부여, 출금 불가 |
| IP 제한 | 거래소 API IP 화이트리스트 |
| 자금 격리 | 봇 전용 서브계정 사용 |
| 로그 보안 | API Key/Secret 마스킹 |