변동성 돌파 전략, 마켓 타이밍 전략

변동성 돌파 전략

특정 조건이 만족되면 매수하고 종가에 매도하는 전략

전일의 변동폭을 이용해 당일의 목표가를 설정하고, 목표가에 도달하면 매수 후 종가에 매도

 

1. 목표가 설정

거래일의 시가에 전일 변동폭의 절반을 더한 값을 목표가로 설정

 

import FinanceDataReader as fdr
import numpy as np

# 1. 코스닥 150 데이터 불러오기 ('229200'은 코스닥 150 ETF의 코드)
df = fdr.DataReader('229200')

# 2. 전일 변동폭 계산 (변동폭: 고가 - 저가)
df['변동'] = df['High'] - df['Low']
df['전일변동'] = df['변동'].shift(1)

 

2. 매수 조건

거래일의 장 중 주가가 목표가를 돌파할 경우 매수

# 3. 목표가 설정 (목표가: 시가 + 전일 변동폭 * 0.5) - 매수조건
df['목표가'] = df['Open'] + df['전일변동'] * 0.5

3. 매도 조건

당일 종가에 매도

 

4. 수익률, 누적 수익률 계산

매수한 후 종가에서 목표가를 나눈 비율로 수익률을 계산하고, 매수가 일어나지 않으면 수익률은 1로 유지

매일의 수익률을 곱해 누적 수익률을 계산

# 거래일의 고가가 목표가보다 크거나 같다면 그 목표가에서 매수
# 만약 매수가 일어나지 않으면 그대로 유지되므로 그 날의 수익률은 1

df['수익률'] = np.where(df['High'] >= df['목표가'], df['Close'] / df['목표가'], 1)
df['누적수익률'] = df['수익률'].cumprod()

1. 수익률 계산:
   • 주식을 매수한 후, 매수한 날의 종가에서 목표가를 나누어 수익률을 계산함.
     • 예를 들어, 매수한 날의 종가가 100원이고 목표가가 110원이라면 수익률은 110/100=1.1 (즉, 10% 수익)
   • 반대로, 목표가보다 종가가 낮다면 수익률이 1보다 작아질 수 있다.
2. 매수가 없을 경우:
   • 특정 날에 매수하는 거래가 발생하지 않았다면, 그날의 수익률은 변동 없이 1로 유지된다는 의미
   • 즉, 그날의 수익률이 그대로 유지되며, 아무런 수익도 손실도 발생하지 않는다는 뜻
3. 누적 수익률 계산:
   • 매일의 수익률을 연속적으로 곱하여 누적 수익률을 계산.
     • 예를 들어, 첫날 수익률이 1.1이고, 둘째 날 수익률이 0.95라면 누적 수익률은 1.1×0.95=1.045
   • 누적 수익률은 시간이 지날수록 매일의 수익률 변동에 따라 계속 업데이트됨.

매수했을 때 목표가 대비 수익률을 계산하고, 매수가 없을 때는 수익률을 유지하며, 누적 수익률은 모든 날의 수익률을 곱해서 계산하는 방식입니다.

 

5. CAGR(연평균 수익률)

주어진 기간 동안의 연평균 수익률을 계산

# 기간 계산 (전체 기간의 일수 계산)
delta = df.index[-1] - df.index[0]  # 시작일과 종료일 사이의 차이 계산
year = delta.days / 365  # 연도 수로 변환

# 연평균 수익률 (CAGR) 계산
CAGR = df['누적수익률'].iloc[-1] ** (1/year) - 1
# 누적 수익률의 가장 마지막 값에서 전체 기간 년도 수로 나누어 복리 수익률 계산
# 수익률을 계산할 때는 1을 빼서 100% 대신 0%부터 시작하도록 맞춰줌 -> 순수한 수익률만 남기게 됨.

print(f"연평균 수익률 (CAGR): {CAGR * 100:.2f}%")

→ 연평균 11.7% 수익률

 

6. 변동성 돌파 전략 vs 단순 보유 비교 시각화

• 변동성 돌파 전략: 이전에 설명한 전략으로, 목표가를 돌파할 때 매수하여 종가에 매도하는 전략
• 단순 보유 전략: 주식을 매수한 후 매매 없이 계속 보유하는 전략입니다. 여기서는 주가의 종가를 기준으로 계산

# 변동성 돌파 전략 > 단순 보유
import matplotlib.pyplot as plt
import koreanize_matplotlib

plt.figure(figsize = (10, 5))
plt.plot(df['누적수익률'])
plt.plot(df['Close'] / df['Close'].iloc[0]) # 종가를 처음 값으로 나누어 비율 계산
plt.grid()
plt.legend(['변동성 돌파 전략', '단순 보유'])
plt.show()

 

7. 최대 낙폭(MDD, Maximum Drawdown) 계산

• MDD: 자산이 가장 큰 고점에서 얼마나 하락했는지를 측정하는 지표
• 현재 시점의 누적 수익률을 그 시점까지의 최고점(전고점)으로 나누어, 자산이 고점 대비 몇 퍼센트인지 계산
• 즉, 최고점에서 최저점까지의 손실률을 의미 ( 자산의 손실 위험을 측정 )

df['전고점'] = df['누적수익률'].cummax()
# 누적된 수익률에서 최고점을 계속 추적

# 현재 시점에서 몇 퍼센트 정도 하락했는지
df['DD'] = (1 - df['누적수익률'] / df['전고점']) * 100
df['DD'].max() # 최대 낙폭 MDD

→ 즉, 자산이 고점에서 가장 크게 하락한 비율

→ 자산의 누적 수익률에서 전고점을 계산하고, 그 전고점 대비 자산이 얼마나 하락했는지를 퍼센트로 나타냄.

---

마켓 타이밍 전략

• 이동평균선(MA) : 주가의 단기나 중기 변동성을 보정해주며, 추세를 확인할 수 있는 지표
• 10일 이동평균선을 기준으로 상승 구간에서만 투자를 하고, 하락 구간에서는 투자를 하지 않는 전략
• 이동 평균선, RSI, 매매 신호 등을 활용하여 투자 기회를 모색하고, XGBoost 모델을 통해 각 ETF의 성과를 예측
• 최종적으로 예측된 종목과 확률을 출력하여 사용자에게 추천

# 마켓 타이밍: 하락 구간일 땐 무시하고 상승 구간에서만 투자
# 10일 이동평균선 위에 있을 때만 투자한다고 가정
df['MA10'] = df['Close'].rolling(window=10).mean()  # 10일 이동 평균선 계산

# 시가가 전날 이동 평균선 위에 있을 때(이동평균선 보다 높을 때) 매매 신호 생성
df['매매신호'] = df['Open'] > df['MA10'].shift(1)  # 시가가 전날(shift(1))의 이동평균선보다 높을 때 매매 신호를 생성
# (df['매매신호'] == 1)는 상승 추세가 이어질 가능성이 높을 때 매수 결정을 하겠다는 의미

# 목표가에 도달했을 때 수익률 계산
df['수익률2'] = np.where((df['매매신호'] == 1) & (df['High'] >= df['목표가']), df['Close'] / df['목표가'], 1)
# (df['매매신호'] == 1): 매매 신호가 발생한 경우(즉, 상승 구간에서만).
# (df['High'] >= df['목표가']): 그날의 최고가가 설정한 목표가에 도달했을 때.
# df['Close'] / df['목표가']: 목표가에 도달한 경우 종가를 목표가로 나누어 수익률을 계산
# 1: 매매 신호가 없거나 목표가에 도달하지 않았을 경우, 수익률은 1(즉, 변동 없음)로 설정

# 누적 수익률 계산
df['누적수익률2'] = df['수익률2'].cumprod()  # 수익률을 누적 곱으로 계산
df['누적수익률2'].iloc[-1]  # 마지막 누적 수익률 확인 (예: 1.56배)

→ 처음 자산 대비 1.56배로 불어났음을 의미합니다(56% 수익).

 

# 최대 낙폭(MDD) : 최대 고점에서 최대 손실까지의 비율 측정 
# (투자 기간 동안 자산의 고점에서 가장 크게 하락한 정도)

df['전고점2'] = df['누적수익률2'].cummax()  # 누적 수익률의 최대값 계산

df['DD2'] = (1 - df['누적수익률2'] / df['전고점2']) * 100  # 최대 손실 계산 (Drawdown)
# (1 - df['누적수익률2'] / df['전고점2']): 현재 누적 수익률이 전고점에서 얼마나 하락했는지 계산
# 0.25가 나온 경우 25% 손실을 의미, *100을 해서 백분율로 표현

df['DD2'].max()  # 최대 손실값 (최대 낙폭) (MDD) 확인

→ 투자 기간 동안 자산의 누적 수익률이 가장 높았던 시점에서 14.36% 하락한 적이 있다

 

---

 

XGBoost 모델을 사용하여 ETF(상장지수펀드)의 수익률을 예측

ETF 데이터로부터 RSI와 가격 변화 등을 사용하여 XGBoost 모델을 학습시키고,

예측된 ETF 종목 중 확률이 0.6 이상인 종목을 추출하여 리스트에 저장

# ETF 수익률 예측 모델링: XGBoost를 사용한 기계 학습 모델
import os
import pandas as pd
from tqdm import tqdm

# RSI: 상대 강도 지수 (0~100)

X = []  # 입력 데이터 (특징, Feature)
Y = []  # 출력 데이터 (라벨, Label)

window = 100  # 윈도우 크기 설정

# ETF 데이터 파일 처리
for i in tqdm(os.listdir("ETFs")):  # "ETFs" 폴더에 있는 파일을 순회
    name = i.split(".")[0]  # 파일 이름에서 확장자를 제외한 이름 추출
    df = pd.read_csv("ETFs/" + i)  # CSV 파일 읽기
    df['Date'] = pd.to_datetime(df['Date'])  # 날짜 열을 datetime 형식으로 변환
    df = df.set_index("Date")  # 날짜를 인덱스로 설정

    data = df[['Change', 'RSI', 'Close']].values  # 필요한 열을 numpy 배열로 변환

    # 윈도우 크기만큼 데이터 슬라이싱
    for j in range(len(data) - window):
        try:
            a = data[j:j + window, :-1].flatten()  # 윈도우 크기만큼 데이터를 평탄화
            b = int(data[j + window - 1, -1] * 1.01 < data[j + window, -1])  # 목표가 도달 여부 (종가가 목표가에 도달했는지 확인)
            # data[j:j + window, :-1]: 현재 인덱스 j부터 윈도우 크기만큼의 데이터(행)와 마지막 열(여기서는 'Close')을 제외한 첫 두 열('Change', 'RSI')을 선택
            # .flatten(): 선택된 2D 배열을 1D 배열로 평탄화. 이렇게 하면 윈도우 크기(100)만큼의 데이터가 한 줄로 나열된 형태가 됨(원래는 행과 열이 있는 2D 배열)
            # 다음 날의 'Close' 값이 목표가의 1% 상승한 값보다 크면 1, 그렇지 않으면 0
        except:
            continue  # 예외 발생 시 다음 반복으로 넘어감
        X.append(a)  # 입력 데이터 추가
        Y.append(b)  # 출력 데이터 추가

from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report
from xgboost import XGBClassifier

# 데이터 분할: 훈련 데이터와 테스트 데이터로 나누기
train_x, test_x, train_y, test_y = train_test_split(X, Y)

# XGBoost 모델 초기화 및 훈련
model = XGBClassifier()
model.fit(train_x, train_y)

# 테스트 데이터에 대한 예측 수행
pred = model.predict(test_x)

# 예측 결과 평가
report = classification_report(test_y, pred)
print(report)

 

total = []  # 예측 결과를 저장할 리스트

# 각 ETF에 대해 최신 데이터를 이용한 예측 수행
for i in os.listdir("ETFs"):
    name = i.split(".")[0]  # 파일 이름에서 주식 이름 추출
    df = pd.read_csv("ETFs/" + i)  # CSV 파일 읽기
    df['Date'] = pd.to_datetime(df['Date'])  # 날짜 열을 datetime 형식으로 변환
    df = df.set_index("Date")  # 날짜를 인덱스로 설정

    data = df[['Change', 'RSI', 'Close']]  # 필요한 열 추출
    today_data = data.iloc[-window:][['Change', 'RSI']].to_numpy().flatten()  # 최근 윈도우 데이터 추출
    try:
        pred = model.predict_proba([today_data])[0][1]  # 확률 예측
        # XGBoost 모델에 today_data를 입력하여 각 클래스(여기서는 목표가 도달 여부 0 또는 1)에 대한 확률을 예측
        # [0][1]: 반환된 배열의 첫 번째 요소(첫 번째 데이터 포인트에 대한 결과)에서 두 번째 값(1에 대한 확률)을 선택
        
        if pred > 0.6:  # 예측 확률이 0.6 이상일 경우
            total.append([name, pred])  # 결과 저장
    except:
        continue  # 예외 발생 시 다음 반복으로 넘어감

total  # 예측된 종목과 확률 리스트 출력

 

from datetime import datetime
today = str(datetime.today().date())

t = f'{today} 기준 종목 예측\n\n'

for i in total:
    t += f'추천종목 : {i[0]} ({i[1]:.2f})\n'

print(t)

 

 

---

LINE Notify를 통한 블로그 글 자동 생성 및 전송

# LINE Notify 패키지 설치
#!pip install line_notify

# LINE Notify를 사용하기 위한 import
from line_notify import LineNotify

# 사용자 고유의 LINE Notify API 키를 설정
my_key = 'b0QYTeG7lM70Os3UkwQZddYRadLweOghfhjgsf4TpU5xAFgzSzcKtzrM'
my_line = LineNotify(my_key)  # LineNotify 객체 생성

# t 변수를 LINE으로 전송
my_line.send(t)  # 변수 t에 저장된 내용을 LINE으로 전송

# OpenAI API를 사용하기 위한 import
from openai import OpenAI

# OpenAI API 키 설정
ai = OpenAI(api_key='sk-isBPj62UYX1yNyH8EBndF6Fbggi0CJovQihtmaeIovT3BlbkFJJsnGigBL5gR-DG8rTwYfZT2CZ9QQI9rujltA-g_NcA')

# OpenAI에게 전달할 메시지를 설정
box = [
    {'role': 'system', 'content': '내가 글을 보여주면 그 글 내용을 블로그에 돌아갈 수 있게 블로그 글로 작성해줘'},  # 시스템 역할 설정
    {'role': 'user', 'content': t}  # 사용자 역할로 변수 t의 내용을 전달
]

# OpenAI GPT 모델을 사용하여 메시지를 생성
model = ai.chat.completions.create(model='gpt-4o', messages=box)

# 생성된 내용을 LINE으로 전송
my_line.send(model.choices[0].message.content)  # 생성된 블로그 글 내용을 LINE으로 전송

# t 변수에 생성된 내용을 저장
t = model.choices[0].message.content

# 생성된 블로그 글 출력
print(t)  # 콘솔에 생성된 내용을 출력

 

 

---

 

 

inf 문제 발생한 경우

import os
import pandas as pd
from tqdm import tqdm

# 데이터 및 레이블을 저장할 리스트
X = []  # 문제지
Y = []  # 정답지

window = 50  # 윈도우 크기

# ETF 파일들을 처리
for i in tqdm(os.listdir("ETFs")):
    name = i.split(".")[0]
    df = pd.read_csv("ETFs/" + i)
    df['Date'] = pd.to_datetime(df['Date'])
    df = df.set_index("Date")

    # 필요한 변수 선택
    data = df[['Change', 'RSI', 'Close']].values  # np.array 형태로 변환

    # 윈도우 기반 데이터 생성
    for j in range(len(data) - window):
        try:
            a = data[j:j + window, :-1].flatten()  # 평탄화
            b = int(data[j + window - 1, -1] * 1.03 < data[j + window, -1])  # 3% 이상 상승 여부
        except:
            continue
        X.append(a)
        Y.append(b)

print("데이터 준비 완료!")

from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report
from xgboost import XGBClassifier

# 데이터 처리
all_data = []
for i in tqdm(os.listdir("ETFs")):
    name = i.split(".")[0]
    df = pd.read_csv("ETFs/" + i)
    df['Date'] = pd.to_datetime(df['Date'])
    df = df.set_index("Date")

    # 파생변수 생성
    df = create_technical_indicators(df)
    
    # 모든 데이터프레임 저장
    all_data.append(df)

X = []
Y = []

window = 50

# X, Y 데이터 생성
for df in all_data:
    data = df[['변동성 지표','전일변동성','RSI', 'Volume Ratio', 'Momentum', 'Close']].values

    for j in range(len(data) - window):
        try:
            a = data[j:j + window].flatten()  # 평탄화

            if np.inf in a:
                continue
            
            b = int(data[j + window - 1, -1] * 1.01 < data[j + window, -1])  # 1% 이상 상승 여부
        except:
            continue
            
        X.append(a)
        Y.append(b)

if np.inf in a:
                continue
→ 이 부분 추가

# 데이터 분할 및 모델 학습
train_x, test_x, train_y, test_y = train_test_split(X, Y, test_size=0.2, random_state=42)

# 모델 학습
model = XGBClassifier()
model.fit(train_x, train_y)

# 성능 평가
pred = model.predict(test_x)
report = classification_report(test_y, pred)
print(report)