# 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`) ```python 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`) ```python 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`) ```python 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`) ```python 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`) ```python 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`) ```python 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`) ```python 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`) ```python 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`) ```python 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`) ```python 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`) ```python @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) ```sql 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`) ```python 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`) ```python 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`) ```python 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 마스킹 |