🧭 퀀트 자동매매 서버 구축 – Flask + Scheduler + DB로 완전 자동화 시스템 만들기

🧭 퀀트 자동매매 서버 구축 – Flask + Scheduler + DB로 완전 자동화 시스템 만들기

지난 글에서는 키움 OpenAPI를 이용한 실시간 자동 매매 코드를 완성했습니다.
이제는 이 로직을 개인 노트북이 아니라, 항상 켜져 있는 서버에서 안정적으로 돌려야 합니다.

오늘은 Python의 Flask 웹 서버, 스케줄러, 데이터베이스를 결합해
실제 펀드 시스템처럼 “자동으로 리밸런싱 → 주문 → 로그 기록 → 대시보드 표시”가 돌아가는
백엔드형 퀀트 매매 엔진을 만들어보겠습니다.

---

1️⃣ 시스템 개요

[ Flask 서버 ]  
     ├── /trade (주문 실행 API)  
     ├── /report (성과 조회 API)  
     ├── /rebalance (팩터 기반 종목 갱신)  
     └── Scheduler (매일 09:00 자동 실행)
          ↓
       [ DB (SQLite or PostgreSQL) ]
          ↓
       [ Log + Report 자동 업데이트 ]

이 구조의 핵심은,
👉 API 서버로 만들어두면 웹/모바일/대시보드 어디서든 제어할 수 있다는 점입니다.

---

2️⃣ 기본 세팅

📦 필수 라이브러리 설치

pip install flask apscheduler pandas sqlalchemy yfinance

• flask: 웹 API 서버
• apscheduler: 주기적 작업 자동 실행
• sqlalchemy: DB ORM
• yfinance: 주가 데이터 수집

---

3️⃣ Flask 서버 기본 구조

from flask import Flask, jsonify
from apscheduler.schedulers.background import BackgroundScheduler
from sqlalchemy import create_engine
import pandas as pd, yfinance as yf, datetime

app = Flask(__name__)
engine = create_engine("sqlite:///quant.db", echo=False)

# 데이터 수집 및 저장 함수
def collect_factor_data():
    tickers = ["005930.KS", "000660.KS", "035420.KS"]
    rows = []
    for t in tickers:
        df = yf.download(t, period="1y")
        ret = (df["Close"].iloc[-1] / df["Close"].iloc[-120]) - 1
        vol = df["Close"].pct_change().std()
        rows.append({"ticker": t, "return": ret, "vol": vol, "updated": datetime.datetime.now()})
    pd.DataFrame(rows).to_sql("factors", engine, if_exists="replace", index=False)
    print("✅ 팩터 데이터 갱신 완료")

# 매매 함수 (단순 예시)
def execute_trade():
    df = pd.read_sql("SELECT * FROM factors", engine)
    top = df.sort_values("return", ascending=False).iloc[0]
    print(f"🚀 {top['ticker']} 매수 실행 (return={top['return']:.2%})")

# 스케줄러 등록
scheduler = BackgroundScheduler()
scheduler.add_job(collect_factor_data, "cron", hour=8, minute=55)  # 데이터 갱신
scheduler.add_job(execute_trade, "cron", hour=9, minute=1)         # 매수 실행
scheduler.start()

@app.route("/report", methods=["GET"])
def report():
    df = pd.read_sql("SELECT * FROM factors", engine)
    return jsonify(df.to_dict(orient="records"))

if __name__ == "__main__":
    collect_factor_data()
    app.run(host="0.0.0.0", port=5000)

---

4️⃣ 코드 설명

반응형

기능 설명

collect_factor_data()	yfinance로 종목 데이터 수집 후 DB 저장
execute_trade()	상위 팩터 점수 종목 자동 매수 (API 연동 가능)
apscheduler	정해진 시각에 함수 자동 실행
/report	현재 포트폴리오 및 팩터 점수 REST API로 제공

💡 이 서버를 AWS EC2, Google Cloud, 또는 NAS 서버에 올려두면,
매일 아침 자동으로 “데이터 수집 → 종목 선택 → 주문 실행”까지 돌아갑니다.

---

5️⃣ DB 구조 (SQLite 예시)

컬럼 설명

ticker	종목 코드
return	최근 6개월 수익률
vol	변동성
updated	업데이트 시각

이 DB는 리포트 생성기(PDF)와 연결되어
매월 자동 투자 성과를 시각화할 수 있습니다.

---

6️⃣ 대시보드 연동 (옵션)

Streamlit 연동 예시

pip install streamlit

import streamlit as st
import pandas as pd
from sqlalchemy import create_engine

engine = create_engine("sqlite:///quant.db")
df = pd.read_sql("SELECT * FROM factors", engine)

st.title("📊 Quant Trading Dashboard")
st.dataframe(df)
st.bar_chart(df.set_index("ticker")["return"])

👉 streamlit run dashboard.py
→ 웹 대시보드에서 실시간으로 팩터 점수와 종목 현황을 확인할 수 있습니다.

---

7️⃣ 운영 자동화 팁

항목 설명

🕓 Scheduler	APScheduler로 정기 실행 (cron 형식)
🧾 로그 관리	logging 모듈로 모든 주문 기록 남기기
☁️ 배포	AWS EC2 / Railway / Docker로 배포
🔒 보안	키움 API 키는 .env에 저장 후 환경변수로 로드
🧠 장애 복구	실패 시 Slack/Email 알림 추가 추천

---

📌 정리

단계 내용

1	Flask 서버로 자동매매 API 구성
2	APScheduler로 자동 실행 등록
3	SQLite로 데이터 영구 저장
4	Streamlit으로 시각화 대시보드 구성
5	클라우드 배포로 완전 자동화 완성

이제 노트북을 켜두지 않아도,
서버가 스스로 투자하고 기록하고 보고하는 시스템이 완성됐습니다.

---

📘 다음 글 예고

다음 편에서는 **“실전 운영 모드 – 오류 감지, 슬랙 알림, 백업 전략”**을 다룹니다.
즉, 실제 운용 환경에서 안정적으로 돌아가게 만드는 운영 자동화(Ops) 구조 설계를 보여드릴게요.

---

 

퀀트서버,Flask자동매매,파이썬스케줄러,apscheduler,quantAPI,데이터투자,자동리밸런싱,퀀트대시보드,streamlit,주식자동화

---