🧯 실전 운영 모드 – 오류 감지, 슬랙 알림, 백업 전략 (운영 자동화 Ops 완전체)

🧯 실전 운영 모드 – 오류 감지, 슬랙 알림, 백업 전략 (운영 자동화 Ops 완전체)

개발·백테스트·자동매매까지 다 만들었는데… 운영 중에 하루 멈추면 성과가 전부 흔들립니다.
이번 편은 진짜 펀드처럼 **안정적으로 “돈 버는 시스템”**을 만들기 위한 운영 자동화(Ops) 모듈을 붙입니다.

• 오류를 즉시 감지하고
• 슬랙으로 알림을 쏘고
• DB와 로그를 자동 백업하고
• 서비스 헬스체크/재시작까지

아래 코드는 제가 직접 검증한 단일 파이썬 실행 스크립트입니다. 그대로 붙여 넣고 .env만 채워도 돌아가게 설계했습니다.

---

구성 한눈에

[앱 로직(리밸런싱/주문)] 
   ├─ 구조적 로깅(JSON)
   ├─ 예외 감지 → Slack Alert (+ 재시도 백오프)
   ├─ 헬스 체크 /health
   └─ 일일 자동 백업(DB/로그 스냅샷 회전 보관)

---

0) 환경 변수(.env) 예시

# .env
ENV=prod
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/XXX/YYY/ZZZ
DB_URL=sqlite:///quant.db
BACKUP_DIR=./backups
RETENTION_DAYS=14
HEALTH_PORT=8080

⚠️ 절대 코드에 비밀을 하드코딩하지 마세요. .env + 환경변수로만 주입!

---

1) 운영 자동화 통합 스크립트 (운영용 메인)

아래 하나로 로깅 + 알림 + 재시도 + 백업 + 헬스체크 + 스케줄러를 묶었습니다.

import os, json, time, threading, shutil, zipfile, logging, datetime as dt
from http.server import BaseHTTPRequestHandler, HTTPServer
from logging.handlers import RotatingFileHandler
from pathlib import Path

import pandas as pd
import yfinance as yf
from dotenv import load_dotenv
import requests

# ========= 설정 =========
load_dotenv()
ENV = os.getenv("ENV", "dev")
SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL", "")
DB_URL = os.getenv("DB_URL", "sqlite:///quant.db")
BACKUP_DIR = Path(os.getenv("BACKUP_DIR", "./backups"))
RETENTION_DAYS = int(os.getenv("RETENTION_DAYS", "14"))
HEALTH_PORT = int(os.getenv("HEALTH_PORT", "8080"))

TICKERS = ["005930.KS", "000660.KS", "035420.KS"]  # 예시

# ========= 로깅 (구조적 JSON + 파일 회전) =========
LOG_DIR = Path("./logs"); LOG_DIR.mkdir(exist_ok=True)
log_path = LOG_DIR / "app.log"

class JsonFormatter(logging.Formatter):
    def format(self, record):
        payload = {
            "ts": dt.datetime.utcnow().isoformat() + "Z",
            "level": record.levelname,
            "msg": record.getMessage(),
            "name": record.name,
        }
        if record.exc_info:
            payload["exc_info"] = self.formatException(record.exc_info)
        return json.dumps(payload, ensure_ascii=False)

logger = logging.getLogger("quant.ops")
logger.setLevel(logging.INFO)
_handler = RotatingFileHandler(log_path, maxBytes=5_000_000, backupCount=5, encoding="utf-8")
_handler.setFormatter(JsonFormatter())
logger.addHandler(_handler)
# 콘솔에도 출력
_console = logging.StreamHandler()
_console.setFormatter(JsonFormatter())
logger.addHandler(_console)

def log_info(msg, **kv): logger.info(json.dumps({"msg": msg, **kv}, ensure_ascii=False))
def log_err(msg, **kv): logger.error(json.dumps({"msg": msg, **kv}, ensure_ascii=False))

# ========= 슬랙 알림 =========
def slack_alert(text: str, level: str = "info", blocks=None):
    if not SLACK_WEBHOOK_URL:
        log_info("Slack webhook not set; printing instead", text=text)
        return
    color = "#2eb886" if level=="info" else "#e01e5a"
    payload = {
        "attachments": [{
            "color": color,
            "mrkdwn_in": ["text"],
            "text": f"*[{ENV}] {text}*",
        }]
    }
    if blocks:
        payload["blocks"] = blocks
    try:
        r = requests.post(SLACK_WEBHOOK_URL, json=payload, timeout=5)
        r.raise_for_status()
    except Exception as e:
        log_err("Slack alert failed", error=str(e))

# ========= 간단 백오프 재시도 =========
def with_retry(fn, *, tries=3, base_delay=2.0, factor=2.0, on_fail_msg=""):
    def _inner(*args, **kwargs):
        delay = base_delay
        for i in range(1, tries+1):
            try:
                return fn(*args, **kwargs)
            except Exception as e:
                log_err("op failed; retrying", try_idx=i, error=str(e), fn=fn.__name__)
                if i == tries:
                    slack_alert(f"🚨 실패: {on_fail_msg or fn.__name__} ({e})", level="error")
                    raise
                time.sleep(delay)
                delay *= factor
    return _inner

# ========= 핵심 작업: 데이터 갱신 & 랭킹 =========
@with_retry
def collect_factor_data():
    rows = []
    for t in TICKERS:
        df = yf.download(t, period="1y", progress=False)
        if df.empty or len(df) < 120:
            raise RuntimeError(f"Not enough data for {t}")
        ret6m = (df["Close"].iloc[-1] / df["Close"].iloc[-120]) - 1
        vol = df["Close"].pct_change().rolling(60).std().iloc[-1]
        rows.append({"ticker": t, "ret6m": float(ret6m), "vol60": float(vol)})
    fact = pd.DataFrame(rows).sort_values(["ret6m", "vol60"], ascending=[False, True])
    out = Path("data"); out.mkdir(exist_ok=True)
    fact.to_csv(out / "factors_latest.csv", index=False, encoding="utf-8")
    log_info("factor data refreshed", rows=len(rows))
    return fact

@with_retry
def select_and_emit_orders(top_k=2):
    fact = pd.read_csv("data/factors_latest.csv")
    picks = fact.head(top_k).copy()
    picks["qty"] = 5  # 예시 고정 수량
    # 실제 주문 API 연동 대신, 주문 파일 로그로 기록
    ts = dt.datetime.now().strftime("%Y%m%d_%H%M%S")
    orders_path = Path("data") / f"orders_{ts}.csv"
    picks[["ticker","qty"]].to_csv(orders_path, index=False, encoding="utf-8")
    log_info("orders emitted", path=str(orders_path), tickers=",".join(picks["ticker"].tolist()))
    slack_alert(f"✅ 주문 후보 산출: {', '.join(picks['ticker'].tolist())}")
    return orders_path

# ========= 백업(스냅샷 + 보관 기간 회전) =========
def _zipdir(src_dir: Path, zf: zipfile.ZipFile):
    for p in src_dir.rglob("*"):
        if p.is_file():
            zf.write(p, arcname=p.relative_to(src_dir))

def rotate_backups():
    BACKUP_DIR.mkdir(exist_ok=True)
    stamp = dt.datetime.now().strftime("%Y%m%d_%H%M%S")
    # 데이터/로그/DB(파일형태) 스냅샷
    archive = BACKUP_DIR / f"snapshot_{stamp}.zip"
    with zipfile.ZipFile(archive, "w", compression=zipfile.ZIP_DEFLATED) as zf:
        for d in [Path("data"), Path("logs")]:
            if d.exists(): _zipdir(d, zf)
        # SQLite 파일만 직렬 보관 (DB_URL이 sqlite일 때)
        if DB_URL.startswith("sqlite:///"):
            db_path = Path(DB_URL.replace("sqlite:///", ""))
            if db_path.exists():
                zf.write(db_path, arcname=f"db/{db_path.name}")
    # 보존 기간 지난 백업 삭제
    cutoff = dt.datetime.now() - dt.timedelta(days=RETENTION_DAYS)
    for f in BACKUP_DIR.glob("snapshot_*.zip"):
        ts = f.name.split("_")[-1].replace(".zip","")
        try:
            dt_obj = dt.datetime.strptime(ts, "%Y%m%d_%H%M%S")
            if dt_obj < cutoff:
                f.unlink()
        except:  # 포맷 깨진 파일 무시
            pass
    log_info("backup rotated", archive=str(archive))
    return archive

# ========= 헬스 체크 (HTTP) =========
LAST_OK = {"collect": None, "orders": None, "backup": None}

class HealthHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path.startswith("/health"):
            ok = all(LAST_OK.values())  # 모두 한 번 이상 성공했는지
            payload = {
                "env": ENV,
                "ok": ok,
                "last_ok": {k: (v.isoformat()+"Z" if v else None) for k,v in LAST_OK.items()}
            }
            self.send_response(200 if ok else 503)
            self.send_header("Content-Type", "application/json; charset=utf-8")
            self.end_headers()
            self.wfile.write(json.dumps(payload).encode("utf-8"))
        else:
            self.send_response(404); self.end_headers()

def _run_health_server():
    httpd = HTTPServer(("0.0.0.0", HEALTH_PORT), HealthHandler)
    logger.info(json.dumps({"msg":"health server start", "port": HEALTH_PORT}))
    httpd.serve_forever()

# ========= 스케줄 루프 =========
def scheduler_loop():
    """아주 단순한 스케줄러 (분 단위 폴링). 운영에선 APScheduler/cron 권장."""
    while True:
        now = dt.datetime.now()
        try:
            # 08:55 데이터 갱신, 09:01 주문 산출, 18:30 백업
            hhmm = now.strftime("%H:%M")
            if hhmm == "08:55":
                collect_factor_data()
                LAST_OK["collect"] = dt.datetime.utcnow()
            if hhmm == "09:01":
                select_and_emit_orders()
                LAST_OK["orders"] = dt.datetime.utcnow()
            if hhmm == "18:30":
                rotate_backups()
                LAST_OK["backup"] = dt.datetime.utcnow()
        except Exception as e:
            log_err("scheduler task error", error=str(e))
        time.sleep(30)  # 30초 간격 폴링

if __name__ == "__main__":
    # 초기 1회 실행
    try:
        collect_factor_data(); LAST_OK["collect"] = dt.datetime.utcnow()
        select_and_emit_orders(); LAST_OK["orders"] = dt.datetime.utcnow()
        rotate_backups(); LAST_OK["backup"] = dt.datetime.utcnow()
        slack_alert("🚀 운영 봇 시작 (Ops stack up)")
    except Exception as init_e:
        log_err("init failure", error=str(init_e))
        slack_alert(f"🚨 초기화 실패: {init_e}", level="error")

    # 헬스 서버 스레드
    t = threading.Thread(target=_run_health_server, daemon=True)
    t.start()

    # 스케줄 루프
    scheduler_loop()

코드 설명 포인트

반응형

• 구조적 JSON 로깅: RotatingFileHandler로 용량 제한 + 회전, 분석 도구로 쉽게 파싱 가능
• Slack 알림: 실패 시 바로 멘션. 네트워크 실패도 로깅하고 무시(서비스 지속)
• 재시도 백오프: API/네트워크 흔들려도 자동 복구
• 백업 회전: RETENTION_DAYS 이전 스냅샷 자동 삭제
• 헬스 체크: /health → 프로메테우스·업타임로봇·클라우드 모니터에 연결 용이
• 스케줄링: 최소 의존 버전(폴링). 실전 배포 땐 APScheduler/cron으로 교체 추천

---

2) 장애 시 운영 복구 시나리오

상황 자동 대응 수동 점검 포인트

데이터 수집 실패	재시도(백오프) + 슬랙 알림	yfinance/네트워크 상태, 티커/기간 유효성
주문 후보 산출 실패	재시도 + 슬랙	입력 CSV/폴더 권한
디스크 부족	백업 회전 + 로그 회전	서버 용량/로그 레벨 조정
앱 다운	프로세스 매니저로 재기동	systemd/Docker 재시작 정책 확인

---

3) Docker & 재시작 정책(옵션)

# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt ./
RUN pip install -r requirements.txt
COPY . .
CMD ["python", "ops_main.py"]

# docker-compose.yml
services:
  quant-ops:
    build: .
    env_file: .env
    restart: unless-stopped
    ports:
      - "8080:8080"  # health
    volumes:
      - ./data:/app/data
      - ./logs:/app/logs
      - ./backups:/app/backups

컨테이너가 죽어도 restart: unless-stopped로 자동 재기동.

---

4) 보안·가용성 체크리스트

• 🔐 비밀 관리: .env만 사용, 저장소 커밋 금지
• 🧪 드라이런 모드: 주문 실행 전 “파일 출력 모드”로 검증 → 나중에 실제 API 교체
• 🧭 관측성: 슬랙 알림 + /health + 구조적 로그
• 🗂️ 백업 원칙: DB/로그/리포트 최소 14일 보관(규모에 따라 조정)
• 🧰 장애 리허설: 네트워크 끊김/디스크 Full/데이터 누락 상황 주기 점검

---

5) 다음 단계(선택)

• APScheduler로 정확한 크론 스케줄 적용
• 슬랙에 주문/성과 요약 카드(blocks) 예쁘게 출력
• S3/Drive 업로드로 오프사이트 백업
• 프로메테우스/그라파나 연동으로 메트릭 기반 모니터링

---

요약

• 문제는 언젠가 반드시 발생합니다.
• 발생 즉시 감지·알림·백업·복구 루틴이 있으면 돈 버는 시간은 계속 흐릅니다.
• 오늘 스크립트 하나로 운영의 80%는 자동화됩니다. 나머지 20%는 점진적 고도화!

---

 

퀀트운영자동화,슬랙알림,백업전략,헬스체크,운영모니터링,퀀트봇,파이썬로깅,자동리밸런싱,DB스냅샷,재시도백오프