박스권 24시간 무중단 그리드 봇 알고리즘

[ 매매 1회 왕복 수수료 계산 ]
매수 수수료: 0.05%
매도 수수료: 0.05%
왕복 합산 수수료 = 0.10%
 
[ 수익 시뮬레이션 ]
만약 그리드 간격을 0.15%로 설정했다면:
순이익 = 0.15% (매매차익) - 0.10% (왕복 수수료) = 단 0.05% (슬리피지 발생 시 실제 손실로 전락!)

def generate_arithmetic_grid(lower_price: float, upper_price: float, grid_count: int):
    """등차 그리드: 가격 간격이 일정 금액(원/USDT)으로 균등 분할"""
    step = (upper_price - lower_price) / grid_count
    return [lower_price + i * step for i in range(grid_count + 1)]
 
def generate_geometric_grid(lower_price: float, upper_price: float, grid_count: int):
    """등비 그리드: 가격 간격이 일정한 백분율(%) 비율로 분할 (장기 추세에 유리)"""
    ratio = (upper_price / lower_price) ** (1.0 / grid_count)
    return [lower_price * (ratio ** i) for i in range(grid_count + 1)]

{
  "symbol": "BTC/USDT",
  "lower_bound": 60000,
  "upper_bound": 70000,
  "last_updated": "2026-10-02T12:00:00Z",
  "levels": [
    { "level": 1, "target_price": 60000, "status": "BUY_FILLED", "order_id": "ord_101" },
    { "level": 2, "target_price": 61000, "status": "SELL_PENDING", "order_id": "ord_102" },
    { "level": 3, "target_price": 62000, "status": "BUY_PENDING", "order_id": "ord_103" }
  ]
}

import json
import logging
from pathlib import Path
from typing import Dict, Any
 
logger = logging.getLogger("GridState")
 
STATE_FILE = Path("/app/data/grid_state.json")
 
class GridStateManager:
    def __init__(self, state_file: Path = STATE_FILE):
        self.state_file = state_file
        self.state_file.parent.mkdir(parents=True, exist_ok=True)
        self.state = self.load_state()
 
    def load_state(self) -> Dict[str, Any]:
        """재부팅 시 디스크에서 이전 그리드 상태 로드"""
        if self.state_file.exists():
            try:
                with open(self.state_file, "r", encoding="utf-8") as f:
                    data = json.load(f)
                    logger.info(f"✅ 이전 그리드 상태를 디스크에서 성공적으로 복원했습니다: {len(data.get('levels', []))}개 층")
                    return data
            except Exception as e:
                logger.error(f"상태 파일 파싱 실패, 신규 생성합니다: {e}")
        
        return {"symbol": "", "levels": []}
 
    def save_state(self, state: Dict[str, Any]):
        """매 체결 시마다 원자적 디스크 쓰기"""
        self.state = state
        temp_file = self.state_file.with_suffix(".tmp")
        # 쓰기 중단으로 파일이 깨지는 것을 방지하기 위해 임시 파일 후 교체
        with open(temp_file, "w", encoding="utf-8") as f:
            json.dump(state, f, indent=2, ensure_ascii=False)
        temp_file.replace(self.state_file)
        logger.debug("그리드 상태가 디스크에 안전하게 동기화되었습니다.")
 
    def update_level_status(self, level: int, status: str, order_id: str):
        for item in self.state.get("levels", []):
            if item["level"] == level:
                item["status"] = status
                item["order_id"] = order_id
                self.save_state(self.state)
                break