인공지능이 파헤친 세력의 숨겨진 패턴, 파이썬 강화학습으로 마스터하는 세력주 공략법!

https://youtu.be/ynl5WsPHodw

안녕하세요, 다빈치스탁의 시청자 여러분. 오늘은 주식 투자의 핵심인 매수와 매도 전략에 대해 깊이 있게 다뤄보려고 합니다. '매수는 과학, 매도는 예술’이라는 증시 격언처럼, 좋은 종목을 고르는 것만큼이나 그 종목을 적절한 시점에 매도하는 것이 역시 매우 중요합니다.

 

그래서 오늘은 인공지능 강화학습을 활용하여 세력의 차트 패턴을 학습하고, 이를 바탕으로 스스로 매매 비중을 조절하는 전략에 대해 알아볼 건데요, 이 방법은 세력의 움직임을 파악하고 그에 따라 투자 결정을 내리는 데 큰 도움이 됩니다.

 

먼저, 강화학습이란 무엇인지 간단히 설명드리자면, 강화학습은 인공지능이 환경과 상호작용하며 보상을 최대화하는 방법을 스스로 학습하는 기술입니다. 주식 시장에서는 가격 변동, 거래량, 차트 패턴 등 다양한 시장 데이터를 환경으로 삼아, 이를 분석하고 최적의 매매 시점을 결정하는 데 사용되는데요.

 

이러한 강화학습으로 세력의 차트 패턴을 반영하여 관측값을 정의하고 매수/매도 전략을 구체화하기 위해, 다음과 같은 단계를 따를 수 있습니다:

 

첫째, 세력의 매집과 급등 패턴을 반영하기 위해, 주가의 이동평균, 거래량, 가격 변동성 등을 포함한 관측값을 정의합니다. 이러한 지표들은 세력의 움직임을 파악하는 데 중요한 역할을 합니다.

 

둘째, 매집구간에서 세력이 주식을 매집하는 패턴을 관찰하여, 매수 시점을 결정합니다. 예를 들어, 역망치형 패턴이 나타나고, 거래량이 증가하며, 주가가 지지선을 유지하는 경우 매수를 고려할 수 있습니다.

 

셋째, 주가가 급등한 후, 천장형 패턴이 나타나거나 거래량이 감소하는 등, 천장형 패턴에서 상단을 돌파하지 못하고 거래량이 감소할 경우 매도를 고려할 수 있습니다.

 

그리고 마지막으로 각 차트 패턴을 구체적으로 정의해줘야 합니다. 예를 들어 천장형 패턴의 경우, 주어진 인덱스로부터 이전 20개의 고가(High) 중 최댓값을 계산합니다. 이어서 계산된 고가들의 표준편차를 구하고, 평균을 구합니다.

 

최종적으로 표준편차가 평균의 5% 미만일 경우 천장형 패턴으로 간주합니다. 즉, 고가들이 매우 근접해 있어 변동성이 낮다는 것을 의미하며, 이는 상승세가 둔화되고 있음을 나타낼 수 있습니다.

 

이 외에도 지지선, 거래량 증가 및 감소, 망치형, 행잉맨 등 여러 가지 조건 역시 구체적으로 정의하여 인공지능이 인식할 수 있는 언어로의 변환이 필요합니다.

 

하지만 이러한 매수와 매도 신호는 때때로 거짓신호를 주기도 합니다. 그래서 “무릎에서 사서 어깨에 팔라”는 말이 나오기도 하죠. 이런 약점을 보완하기 위해 분할매수와 분할매도로 접근하도록 인공지능을 설계해 보았습니다.

 

보시는 것처럼 random choice 부분은 확률 분포(buy probabilities)에 따라 5%, 10%, 15% 중 하나를 선택합니다. 이는 강화학습에서 에이전트가 특정 행동을 취할 확률을 결정하는 데 사용됩니다.

 

초기에는 buy weights가 임의로 설정되어 있지만, 강화학습 과정에서 에이전트는 보상을 최대화하는 행동을 학습합니다. 이 과정에서 에이전트는 실험을 통해 어떤 매수 비율이 가장 좋은 결과를 가져오는지를 배우고, 이를 조정하여 시간이 지남에 따라 최적의 매수 비율을 선택하는 확률 분포를 학습하게 됩니다.

 

그럼 지금부터 최근 핫한 종목으로 떠오르고 있는 가온전선에 대해 실전 테스트를 진행해보겠습니다. 23년 9월부터 오늘까지 최근 6개월 데이터를 사용하였고 총 1만 번의 학습을 거쳐 주가 움직임과 보유비중을 시각화하였습니다.

 

테스트 결과 거래량이 증가하는 상승의 초입에는 매수비중을 늘리고 추세가 둔화되는 시점이나 천장형 패턴에서는 비중을 축소하여 현금을 늘리는 전략을 취하고 있습니다. 이렇게 인공지능을 활용하면 자칫 놓치기 쉬운 시장의 시그널을 포착하여 보다 냉철한 판단을 내리는 데 큰 도움이 됩니다.

 

여러분들도 제공된 소스코드를 활용하여 자신만의 매매 전략을 개발하고, 시장의 움직임에 더욱 유연하게 대처해보세요. 오늘의 강의가 투자에 도움이 되었기를 바라며, 다음 시간에 더 유익한 내용으로 찾아뵙겠습니다. 시청해주셔서 감사합니다.

 

세력의 차트 패턴 파이썬 강화학습 소스코드

import numpy as np
import gym
from gym.spaces import Box, Discrete
import numpy as np

def softmax(x):
    """Compute softmax values for each set of scores in x."""
    e_x = np.exp(x - np.max(x))
    return e_x / e_x.sum(axis=0)  # only difference

def is_morning_star(candles, index):
    # 모닝스타 패턴은 3개의 캔들로 구성됩니다.
    if index < 2:  # 최소 3개의 캔들이 필요합니다.
        return False
    first, second, third = candles.iloc[index-2], candles.iloc[index-1], candles.iloc[index]
    # 첫 번째 캔들은 음봉, 두 번째 캔들은 작은 실체, 세 번째 캔들은 양봉이어야 합니다.
    return (first['Close'] < first['Open'] and
            abs(second['Close'] - second['Open']) < abs(first['Close'] - first['Open']) * 0.5 and
            third['Close'] > third['Open'] and
            third['Close'] > first['Close'])

def is_evening_star(candles, index):
    # 이브닝스타 패턴은 3개의 캔들로 구성됩니다.
    if index < 2:  # 최소 3개의 캔들이 필요합니다.
        return False
    first, second, third = candles.iloc[index-2], candles.iloc[index-1], candles.iloc[index]
    # 첫 번째 캔들은 양봉, 두 번째 캔들은 작은 실체, 세 번째 캔들은 음봉이어야 합니다.
    return (first['Close'] > first['Open'] and
            abs(second['Close'] - second['Open']) < abs(first['Open'] - first['Close']) * 0.5 and
            third['Close'] < third['Open'] and
            third['Close'] < first['Close'])

def is_hanging_man(candles, index):
    # 행잉맨 패턴은 상승 추세에서 나타나는 장대음봉입니다.
    if index < 1:  # 최소 2개의 캔들이 필요합니다.
        return False
    last_candle = candles.iloc[index]
    body_size = last_candle['Open'] - last_candle['Close']
    lower_wick = last_candle['Close'] - last_candle['Low']
    upper_wick = last_candle['High'] - max(last_candle['Open'], last_candle['Close'])
    # 몸통은 작고, 아래 꼬리는 몸통보다 길어야 합니다.
    return (body_size > 0 and
            lower_wick >= 2 * body_size and
            upper_wick < body_size)

def is_inverted_hammer(candles, index):
    # 역망치 패턴은 하락 추세에서 추세반전을 알리며 나타나는 장대양봉입니다.
    if index < 1:  # 최소 2개의 캔들이 필요합니다.
        return False
    last_candle = candles.iloc[index]
    body_size = last_candle['Close'] - last_candle['Open']
    lower_wick = last_candle['Open'] - last_candle['Low']
    upper_wick = last_candle['High'] - max(last_candle['Open'], last_candle['Close'])
    # 몸통은 작고, 위 꼬리는 몸통보다 길어야 합니다.
    return (body_size > 0 and
            upper_wick >= 2 * body_size and
            lower_wick < body_size)

def volume_increase(data, index, threshold=1.5):
    if index < 20:  # 최소 20개의 캔들이 필요합니다.
        return False
    # 거래량 증가 인식 로직
    # 현재 거래량이 이전 거래량의 평균보다 특정 비율 이상 증가했는지 확인
    current_volume = data['Volume'].iloc[index]
    average_volume = data['Volume'].rolling(window=20, min_periods=1).mean().iloc[index]
    if current_volume > (average_volume * threshold):
        return True
    return False

def calc_price_support(data, index):
    if index < 20:  # 최소 20개의 캔들이 필요합니다.
        return False
    # 가격 지지 인식 로직
    # 최근20일간 거래량 가중 평균 가격을 계산
    vwap = (data['Volume'][index-19:index + 1] * data['Close'][index-19:index + 1]).sum() / data['Volume'][index-19:index + 1].sum()
    current_price = data['Close'].iloc[index]
    # 현재 가격이 VWAP 이상인지 확인
    return current_price >= vwap

def price_support(data, index, support_level):
    # 가격 지지 인식 로직
    # 현재 가격이 지지선 이상인지 확인
    current_price = data['Close'].iloc[index]
    if current_price >= support_level:
        return True
    return False

def ceiling_pattern(data, index):
    if index < 20:  # 최소 20개의 캔들이 필요합니다.
        return False
    # 천장형 패턴 인식 로직 : 최근 고점들이 비슷한 수준에 형성되었는지 확인
    # 지정된 인덱스 범위 내에서 각각의 20일 고가(High)의 최댓값들의 시리즈를 반환
    recent_highs = data['High'].rolling(window=20, min_periods=1).max().iloc[index-19:index+1]

    # 표준편가 평균의 5% 미만일 경우 천장형 패턴으로 간주
    # 신고가를 갱신하지 못하고 이전 고점을 쉽게 돌파하지 못하는 형국
    if recent_highs.std() < (recent_highs.mean() * 0.05):
        return True
    return False

def volume_decrease(data, index, threshold=0.5):
    if index < 20:  # 최소 20개의 캔들이 필요합니다.
        return False
    # 거래량 감소 인식 로직
    # 현재 거래량이 이전 거래량의 평균보다 특정 비율 이하로 감소했는지 확인
    current_volume = data['Volume'].iloc[index]
    average_volume = data['Volume'].rolling(window=20, min_periods=1).mean().iloc[index]
    if current_volume < (average_volume * threshold):
        return True
    return False

# 주식 투자 환경 정의하기 
class StockEnv(gym.Env):   
    def __init__(self, df):
        self.data = df
        self.current_step = 0
        self.portfolio_value = 1000000  # 초기 투자 금액
        self.initial_account_balance = self.portfolio_value
        self.holdings = 0  # 보유 주식 수
        self.cash = self.portfolio_value  # 현금 잔고
        self.action_space = Discrete(3)  # 가능한 행동: 매수, 매도, 보류
        self.observation_space = Box(low=-np.inf, high=np.inf, shape=(10,))  # 관측값의 범위와 형태
        self.reset()
                
    def _get_observation(self):
        # 현재 스텝의 주가 정보 : 세력의 캔들패턴을 추정하기 위한 관측자료
        current_open = self.data['Open'].iloc[self.current_step]
        current_High = self.data['High'].iloc[self.current_step]
        current_Low = self.data['Low'].iloc[self.current_step]
        current_price = self.data['Close'].iloc[self.current_step]
        # 이동평균 : 세력의 장기 추세선을 120일 이평으로 가정, 250일 혹은 기타 장기추세로 전환가능
        moving_average = self.data['Close'].rolling(window=120, min_periods=1).mean().iloc[self.current_step]
        # 거래량 : 세력 종목의 가장 큰 특징인 거래량 변동을 관측값으로 전달
        volume = self.data['Volume'].iloc[self.current_step]
        # 가격 변동성 : 
        volatility = self.data['Close'].rolling(window=20, min_periods=1).std().fillna(0).iloc[self.current_step] 
        # 매집구간 동안의 평균 거래량 : 세력의 매집기간을 6개월로 계산, 실제 3개월 ~ 수 년까지 다양한 패턴
        avg_volume_during_accumulation = self.data['Volume'].rolling(window=120, min_periods=1).mean().iloc[self.current_step]
        # 가격의 최저점 : 지지라인을 계산하기 위한 저점을 관측값으로 전달
        low_price_during_accumulation = self.data['Low'].rolling(window=120, min_periods=1).min().iloc[self.current_step]
        # 가격의 최고점 : 저항라인을 계산하기 위한 고점을 관측값으로 전달
        high_price_during_accumulation = self.data['High'].rolling(window=120, min_periods=1).max().iloc[self.current_step]
        
        # 관측값 배열에 추가 지표들 포함
        observation = np.array([
            current_open, 
            current_High, 
            current_Low, 
            current_price, 
            moving_average, 
            volume, 
            volatility, 
            avg_volume_during_accumulation, 
            low_price_during_accumulation, 
            high_price_during_accumulation
        ])
        return observation
    
    def reset(self):
        # 초기 상태 설정
        self.current_step = 0 # 일자별 단계
        self.holdings = 0 # 주식보유수량
        self.portfolio_value = 1000000 # 초기 투자금액
        self.initial_account_balance = self.portfolio_value # 수익률 계산을 위한 시작금액
        self.cash = self.portfolio_value # 주식을 팔아서 가지고 있는 현금 비중
        return self._get_observation()  # 초기 관측값 반환

    # 매수 전략 구체화 : 역망치, 모닝스타, 지지선유지, 거래량 증가
    def buy_strategy(self):
        # 지지라인 계산 : 세력의 매집원가 계산
        support_level = calc_price_support(self.data, self.current_step)
        # 역해머형 캔들이 발생하고 가격이 지지라인 위에서 놀고 있으며 거래량이 점진적으로 증가하는 시점
        if is_inverted_hammer(self.data, self.current_step) and price_support(self.data, self.current_step, support_level) and volume_increase(self.data, self.current_step):
            return True
        # 모닝스타 캔들이 발생하고 가격이 지지라인 위에서 놀고 있으며 거래량이 점진적으로 증가하는 시점
        elif is_morning_star(self.data, self.current_step) and price_support(self.data, self.current_step, support_level) and volume_increase(self.data, self.current_step):
            return True
        else:
            return False

    # 매도 전략 구체화 : 천장형, 행잉맨, 이브닝스타, 거래량 감소, 지지선 이탈
    def sell_strategy(self):
        # 지지라인 계산 : 세력의 매집원가 계산
        support_level = calc_price_support(self.data, self.current_step)
        # 천장형 차트 패턴이 발생하고 거래량이 점진적으로 감소하는 시점
        if ceiling_pattern(self.data, self.current_step) and volume_decrease(self.data, self.current_step):
            return True
        # 교수형 차트 패턴이 발생하고 거래량이 점진적으로 감소하는 시점
        elif is_hanging_man(self.data, self.current_step) and volume_decrease(self.data, self.current_step):
            return True
        # 석별형 차트 패턴이 발생하고 거래량이 점진적으로 감소하는 시점
        elif is_evening_star(self.data, self.current_step) and volume_decrease(self.data, self.current_step):
            return True
        # 주가가 지지라인을 이탈하는 시점
        elif price_support(self.data, self.current_step, support_level) == False:
            return True
        else:
            return False
            
    def step(self, action):
        # 행동 실행 및 상태 업데이트
        current_price = self.data['Close'].iloc[self.current_step]
        reward = 0
        done = False
        info = {}

        if action == 0:  # 매수
            if self.buy_strategy():
                # 매수 비율을 결정하는 로직 (Softmax 함수 활용)
                buy_weights = np.array([0.1, 0.2, 0.3])  # 분할 매수 가중치 예시
                buy_probabilities = softmax(buy_weights) # 최적화 된 확률분포 학습
                # 분할 매수 비중 예시 : 공격적인 사용자의 경우 다른 비중으로 교체가능
                buy_percentage = np.random.choice([5, 10, 15], p=buy_probabilities)
                buy_amount = self.cash * (buy_percentage / 100)
                self.holdings += buy_amount / current_price
                self.cash -= buy_amount
            # ...
        elif action == 1:  # 매도
            if self.sell_strategy():
                # 매도 비율을 결정하는 로직 (Softmax 함수 활용)
                sell_weights = np.array([0.1, 0.2, 0.3])  # 분할 매도 가중치 예시
                sell_probabilities = softmax(sell_weights) # 최적화 된 확률분포 학습
                # 분할 매도 비중 예시 : 공격적인 사용자의 경우 다른 비중으로 교체가능
                sell_percentage = np.random.choice([5, 10, 15], p=sell_probabilities)
                sell_amount = self.holdings * (sell_percentage / 100)
                self.cash += sell_amount * current_price
                self.holdings -= sell_amount
                #print("매도")

        # 주식 가격 변화에 따른 보상 계산
        prev_price = self.data['Close'].iloc[self.current_step - 1]
        price_change = current_price - prev_price
        reward += price_change * self.holdings

        # 종료 여부 확인 (예: 최대 스텝 도달)
        done = self.current_step >= len(self.data) - 1
        if not done:
            self.current_step += 1
            new_state = new_state = self._get_observation() if not done else np.zeros(self.observation_space.shape)
        else:
            new_state = [0, 0, 0, 0, 0, 0, 0]  # 종료 시 적절한 기본값 설정
        
        info.update({'balance': self.cash, 'portfolio': self.holdings, 'reward': reward})

        return new_state, reward, done, info

    def render(self, mode='human'):
        # 현재 포트폴리오 가치 계산
        current_portfolio_value = self.cash + self.holdings * self.data['Close'].iloc[self.current_step]
        # 초기 투자 대비 수익률 계산
        profit = current_portfolio_value - self.initial_account_balance
        profit_percentage = (profit / self.initial_account_balance) * 100

        # 출력
        print(f"Total Days: {self.current_step}")
        print(f"Current Portfolio Value: {current_portfolio_value:.2f}")
        print(f"Profit: {profit:.2f} ({profit_percentage:.2f}%)")

import yfinance as yf
from stable_baselines3 import PPO, A2C
from stable_baselines3.common.vec_env import DummyVecEnv
from stable_baselines3.common.evaluation import evaluate_policy

df = yf.download("000500.KS", start="2023-09-16", end="2024-04-21")
df = df.dropna()
# 환경 생성
env = StockEnv(df)
"""
# 모델 생성
model = A2C("MlpPolicy", env, verbose=1)
# 모델 학습
model.learn(total_timesteps=10000)
# 모델 평가
mean_reward, std_reward = evaluate_policy(model, env, n_eval_episodes=10)
print(mean_reward, std_reward)
# 학습된 모델 저장
model.save("gaon_electronics")
"""
# 저장된 모델 불러오기
model = A2C.load("gaon_electronics", env=env)

import matplotlib.pyplot as plt

# 포트폴리오 가치와 보유 주식 수량 기록을 위한 리스트 초기화
holings_list = []
close_list = []

# 학습된 모델로 환경 시뮬레이션
obs = env.reset()
while True:
    action, _states = model.predict(obs, deterministic=True)
    obs, rewards, dones, info = env.step(action)
    
    # 현재 포트폴리오 가치 및 보유 주식 수량 계산 및 기록
    current_portfolio_value = env.cash + env.holdings * env.data['Close'].iloc[env.current_step]
    holings_list.append(env.holdings)
    close_list.append(env.data['Close'].iloc[env.current_step])
    
    if dones:
        env.render()
        break

# 포트폴리오 가치 그래프 그리기
fig, ax1 = plt.subplots(figsize=(10, 6))

color = 'tab:blue'
ax1.set_xlabel('Time Steps')
ax1.set_ylabel('Holdings Quantity', color=color)
ax1.plot(holings_list, label='Holdings Quantity', color=color)
ax1.tick_params(axis='y', labelcolor='black')

# 보유 주식 수량 그래프 추가 (두 번째 y축 사용)
ax2 = ax1.twinx()  # 같은 x축을 공유하는 두 번째 y축 생성
color = 'tab:red'
ax2.set_ylabel('Stock Price', color=color)
ax2.plot(close_list, label='Stock Price', color=color, linestyle='--')
ax2.tick_params(axis='y', labelcolor='black')

fig.tight_layout()  # 그래프 레이아웃 조정
plt.title('Stock Price and Holdings Quantity Over Time')
fig.legend(loc="upper left", bbox_to_anchor=(0.1, 0.9))
plt.show()