반응형
미국의 수학자 에드워드 소프가 맥스웰-볼츠만 분포로 퀀트 알고리즘을 만들어 대박을 냈다는 뉴스를 봤습니다.
지피티와 함께라면 나도 할 수 있다라는 생각에 만들어 보기로 했어요.
[맥스웰-볼츠만 분포]
물리학에서 사용되는 이론인데, 공기역학에서 입자들의 에너지는 평균으로 수렴하고 분포 양 끝에 있는 움직임은 이상치라는 이론입니다.
[투자에 어떻게 적용?]
기업의 주가 또한 평균적인 움직임 내에서 움직이며 상승/하락을 하는 것이고, 만약 평균 밖의 상승/하락이 있다면 이상치이기 때문에 매수/매도하는 전략입니다.
이 전략을 적용하기 위해서는 '평균'을 낼 수 있어야 하므로, 백데이터가 많고, 거래량이 많고, 시장의 참여자가 많은 종목이어야 확률이 올라갑니다.
[코드 개발]
1. 파이썬 무료 라이브러리인 야후의 yfinance를 사용하여 백데이터를 얻는다.
2. 하위 20%와 상위 10%의 움직임을 이상치라고 정의한다.
3. 매일 밤 10시 25분(한국시간)에 미국 주식 전날 종가 기준으로 해당 종가의 움직임(%)이 이상치에 해당하는지 판별한다.
4. 메일로 발송하여 매수/매도를 잊지 않도록 한다.
아주 단순하죠?
그럼 코드를 보겠습니다.
역시나 지피티가 열심히 만들어주었습니다.
# MB 분포 기반 퀀트 시스템 예제 (목표: 월 수익률 3%, 연 30%) - 매수/매도 신호 유지형 전략 + 손절/익절 (5% 이익, -1.5% 손절)
# 초기 모델
# 매수, 매도 시그널과 익절/손절 조건이 구현되어 있지만 백테스트 로직에서 매수 후 다음날 매도에 대한 수익률만 계산함.
# 즉, 백테스트 수익률과 실제 모델의 로직 결과가 다를 수 있음.
# 불안정한 모델.
import pandas as pd
import numpy as np
from scipy.stats import maxwell
import yfinance as yf
from datetime import datetime
from dotenv import load_dotenv
import os
from module.mail import send_email # ← 분리된 메일 함수 불러오기
# === 환경 변수 불러오기 ===
load_dotenv()
sender_email = os.getenv("EMAIL_SENDER")
recipient_emails = os.getenv("EMAIL_RECIPIENTS", "").split(',')
smtp_server = os.getenv("SMTP_SERVER")
smtp_port = int(os.getenv("SMTP_PORT"))
smtp_user = os.getenv("SMTP_USER")
smtp_password = os.getenv("SMTP_PASSWORD")
# === 1. 데이터 불러오기 (예: AAPL) ===
ticker = 'TSLA'
today = datetime.today().strftime('%Y-%m-%d')
df = yf.download(ticker, start='2025-01-01', end=today) # auto_adjust=True by default
info = yf.Ticker(ticker).info
company_name = info.get("shortName", ticker) # ▶ 회사명 가져오기 (없으면 티커 그대로)
df['return'] = df['Close'].pct_change()
df.dropna(inplace=True)
# === 2. MB 분포 시그널 생성 ===
rolling_window = 15 # n일 기준 변동성
scale = df['return'].rolling(rolling_window).std()
df['mb_pdf'] = maxwell.pdf(np.abs(df['return']), scale=scale)
# 이상치 시그널: 하위 10%는 매수 조건, 상위 10%는 매도 조건
lower_q = df['mb_pdf'].rolling(rolling_window).quantile(0.2) # 매수(급락) 시그널
upper_q = df['mb_pdf'].rolling(rolling_window).quantile(0.9) # 매도(급등) 시그널
df['signal'] = 0
# ▶ 매수 시그널
df.loc[df['mb_pdf'] < lower_q, 'signal'] = 1
# ▶ 매도 시그널
df.loc[df['mb_pdf'] > upper_q, 'signal'] = -1
# === 3. 포지션 유지 로직 구현 ===
df['position'] = df['signal'].replace(to_replace=0, method='ffill').fillna(0)
# === 4. 시가 기준 전략 수익률 계산 (포지션 유지 + 손절/익절 적용) ===
df['current_open'] = df['Open']
df['next_open'] = df['Open'].shift(-1)
df['raw_return'] = (df['next_open'] - df['current_open']) / df['current_open']
# 전날 대비 하락/상승시 매도/매수 기준
stop_loss = -0.015
take_profit = 0.1
# 수익률 클리핑 적용 (포지션 있는 날에만 적용)
df['strategy_return'] = 0.0
active = df['position'] != 0
df.loc[active, 'strategy_return'] = df.loc[active, 'raw_return'].clip(lower=stop_loss, upper=take_profit)
# 누적 수익률 계산
df['cumulative_strategy'] = (1 + df['strategy_return']).cumprod()
df['cumulative_market'] = (1 + df['return']).cumprod()
# === 5. 성과 평가 ===
def evaluate_strategy(df):
monthly_return = df['strategy_return'].resample('ME').sum()
mean_monthly = monthly_return.mean()
annual_return = mean_monthly * 12
std_monthly = monthly_return.std()
sharpe = (mean_monthly / std_monthly) * np.sqrt(12)
return mean_monthly, annual_return, sharpe
monthly, annual, sharpe = evaluate_strategy(df)
print(f"[MB 전략 성과] \n월 수익률: {monthly*100:.2f}%\n연 수익률: {annual*100:.2f}%\n샤프비율: {sharpe:.2f}")
# === 6. 시그널 요약 ===
print("\n신호 발생 요약:")
signal_counts = df['signal'].value_counts()
for signal_value, count in signal_counts.items():
if signal_value == 1:
label = '급락 시그널'
elif signal_value == -1:
label = '급등 시그널'
else:
label = '유지/무신호'
print(f"{label} ({signal_value}): {count}건")
# 오늘 날짜 기준 (가장 최근 row) 확인
today_signal = df['signal'].iloc[-1]
today_position = df['position'].iloc[-1]
# === 7. 오늘 매수 여부 판단 ===
latest_date = df.index[-1].strftime('%Y-%m-%d')
message_body = ""
if today_signal == 1:
msg = f"✅ {latest_date}: 급락 시그널 발생 → 오늘 줍줍하거나 손절하세요."
print(msg)
message_body += msg
elif today_signal == -1:
msg = f"🚨 {latest_date}: 급등 시그널 발생 → 오늘 차익 실현 고려! 🚀"
print(msg)
message_body += msg
elif today_position == 1:
msg = f"❌ {latest_date}: 포지션 유지 중 → 보유 지속"
print("📈 포지션 유지 중 → 보유 지속")
message_body += msg
else:
msg = f"❌ {latest_date}: 시그널 없음 → 오늘은 관망하세요."
print(msg)
message_body += msg
# === 9. 오늘 손절 조건 확인 ===
if today_position == 1 and df['raw_return'].iloc[-1] <= stop_loss:
msg = f"⚠️ {stop_loss} 이상 하락 → 손절 고려하세요."
print(msg)
message_body += msg
# === 이메일 전송 실행 (다중 수신자) ===
for recipient_email in recipient_emails:
send_email(
subject = f"[MB 퀀트] {latest_date} 시그널 - {company_name}",
body = message_body,
sender = sender_email,
recipient = recipient_email.strip(),
smtp_server = smtp_server,
smtp_port = smtp_port,
smtp_user = smtp_user,
smtp_password = smtp_password
)
위 코드는 테슬라를 2025년 1월 1일부터 오늘자 데이터까지를 바탕으로 맥스웰-볼츠만 분포 모델로 분석한 것입니다.
이동평균선 일수, 이상치 기준, 손절/익절 기준 등..은 여러가지 넣어보면서 테스트해야 합니다.
코드를 실행하면 아래처럼 출력되는데요.
진짜 이대로만 수익이 난다면 누구나 다 떼부자가 됐겠죠? 대출을 받아서라도 전재산을 투자했을거에요.
몇시간동안 여러 변수를 넣어 돌려보고, 한 달 동안 모니터링하며 실제 1주씩 매수매도 해본 결과...
엄청나게 잘 맞지는 않습니다.
저 위의 알고리즘 코드는 초기버전으로, 저는 지피티와 열심히 티격태격하며 버전을 계속 업그레이드 했습니다.
단순히 BM 모델을 적용할 뿐 아니라, 백데이터로 테스트하는 과정에서 신경써야할 조건들이 있었기 때문입니다.
홀딩할 시 가격의 평가 기준, 매수/매도의 이상치 기준 등등...
그래도 이렇게 매일 밤 메일로 받으니, 전날 움직임을 알 수 있고 확실히 좋긴 합니다.
아래는 메일을 보내는 코드입니다.
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
def send_email(subject, body, sender, recipient, smtp_server, smtp_port, smtp_user, smtp_password):
msg = MIMEMultipart()
msg['From'] = sender
msg['To'] = recipient
msg['Subject'] = subject
msg.attach(MIMEText(body, 'plain'))
try:
with smtplib.SMTP_SSL(smtp_server, smtp_port) as server:
server.login(smtp_user, smtp_password)
server.send_message(msg)
print("📧 이메일 전송 완료!")
except Exception as e:
print("이메일 전송 실패:", e)
반응형