[Python] Enum 클래스를 이용한 일반화 작업

개인 프로젝트를 하던 도중에 여러 분석 기법들을 이용해 데이터를 분석하는 코드를 작성해야 했다.

 

[처음 도입한 방식 : 공통 테이블]

 

추후 추가될 가능성을 생각해 코드의 일반화가 필요했고, 하드코딩을 줄이고 순수 로직만을 코드에 녹여내기 위해서 분석 기법을 관리하는 테이블을 생성하기로 결정했다. 설계한 테이블은 아래와 같다.

 

분석기법 마스터 테이블

 

trade_cls_cd	분석기법코드
trade_cls_nm	분석기법명
timing_period	분석기간
safe_degree	기법 신뢰도
batch_name	분석 배치명
use_yn	사용유무

 

해당 테이블을 배치에서 조회해 for을 돌리는 방식으로 구성해, 최대한 수정을 줄일 수 있는 방향으로 코드를 작성하고자 했다.

(몇 가지 기법을 나중에 추가해도 테이블에 데이터만 넣으면 된다.)

 

 

배치 파일, TimingDayNotify에서 AnalysisDayXX를 하나씩 호출한다.

 

 

- timingDayNotify.py(분석 배치, 하나씩 분석기법들을 호출한다.)

생략 ...

# 1. 투자기법 조회
sql = f"""
	SELECT *
	FROM MA_TRADE_WAY
	WHERE use_yn = 'Y'
"""

analysis_list = self.engine.execute(sql).fetchall()

# 2. FROM_DATE 구하기 --> 이 수치 기반으로 ST_PRICE_DAY 조회
sql = f"""
	SELECT MAX(timing_period)
	FROM MA_TRADE_WAY
	WHERE use_yn = 'Y'

"""

max_period = self.engine.execute(sql).fetchone()
max_period = max_period[0]
from_date = (datetime.now().date() - relativedelta(months=max_period)).strftime('%Y-%m-%d')

# 3. 한 종목씩 DB업데이트 호출
for idx, stock in self.stock_list.iterrows():
    sql = f"""
        SELECT *
        FROM ST_PRICE_DAY
        WHERE market_cd = '{stock.market_cd}'
    AND stock_cd = '{stock.stock_cd}'
    AND stock_date >= '{from_date}'
    """
    df_prices = pd.read_sql(sql, con=self.engine)
    for al in analysis_list:
        trade_way = al[4]
    	analysis_period = al[2]
        # 실제 구하려는 기간도 파라미터로 넘긴다.
    	rs_prices = self.analysis_day_price(df_prices, trade_way, analysis_period) # <---이 메서드에서 분석 배치들을 호출한다.
    	if rs_prices is None:
    		continue
    print('[{}] #{:06d} {} ({}) : {} rows > ANALYSIS ST_PRICE_DAY [OK]'.format(datetime.now().strftime('%Y-%m-%d %H:%M'), idx+1, stock['stock_cd'], stock['market_cd'], len(rs_prices)))
print('-------------FINISH UPDATE ST_PRICE_TIME-------------- ')

... 생략

배치 일부를 발췌한 코드이다. MA_TRADE_WAY(분석기법 마스터)에 저장된 분석 기법들을 조회해 테이블 row수만큼 for을 이용해 저장된 배치(ex. AnalysisDay10, AnalysisDay11)를 호출한다. 

 

- AnalysisDay10.py(실제 분석 기법이 담겨있다.)

class AnalysisDay10:
    def __init__(self, df_input, period):
        print("CHECK-----")
        self.df_prices = df_input
        self.period = period

    def get_analysis_result(self):

        df_prices = self.df_prices
        period = self.period

        # 1.볼린저 밴드
        df_prices['MA20'] = df_prices['close'].rolling(window=20).mean()
        df_prices['stddev'] = df_prices['close'].rolling(window=20).std()
        df_prices['upper'] = df_prices['MA20'] + (df_prices['stddev'] * 2)
        df_prices['lower'] = df_prices['MA20'] - (df_prices['stddev'] * 2)
        ...생략

 

위 두 코드는 마스터 테이블(MA_TRADE_WAY)을 조회해 테이블에 저장된 배치(ex. AnalysisDay10)들을 순서대로 호출하여 분석결과를 새로운 테이블(ST_TRADE_CHECK)에 삽입하는 구조다.

 

여기서 문제는 데이터를 코드에 녹여내지 않고 테이블로 관리하기 때문에 1. 데이터의 히스토리 관리가 불가하고(누가 언제 데이터를 삽입했는지 모른다), 2. 관리의 주체가 객체가 아닌 DB라는 점에서, 그리고 테이블이 아닌 Enum클래스를 파이썬에도 사용할 수 없을까라는 3. 호기심으로 Enum클래스를 한 번 적용해보기로 했다.

---

[관리의 주체를 DB에서 객체로: Enum클래스]

 

1. 우선 Enum클래스를 설계했고,(테이블에 저장된 데이터 기반으로 코드를 작성했다.)

from enum import Enum

class Analysis(Enum):
    TRADE_FOLLOWING = (10, 5, 50, 'AnalysisDay10', 'Y')
    REVERSALS       = (11, 5, 50, 'AnalysisDay11', 'Y')
    TRIPLE_SCREEN   = (20, 5, 70, 'AnalysisDay20', 'Y')
    DUAL_MOMENTUM   = (30, 5, 90, 'AnalysisDay30', 'Y')

    def __init__(self, trade_cls_cd, timing_period, safe_degree, batch_name, use_yn):
        self.trade_cls_cd = trade_cls_cd
        self.timing_period = timing_period
        self.safe_degree = safe_degree
        self.batch_name = batch_name
        self.use_yn = use_yn

    def get_batch_name(self):
        return self.batch_name

    def get_timing_period(self):
        return self.timing_period

    def get_use_yn(self):
        return self.use_yn

코드 내 사용할 Enum변수와 초기화, 필요한 함수를 작성했다.

 

2. 이 Enum 클래스를 바탕으로 코드를 재작성했다.

# 1. 분석기간 가장 큰 기간 구하기 --> 이 수치 기반으로 ST_PRICE_DAY 조회
        period_list = [a.get_timing_period() for a in Analysis]
        max_period = max(period_list)
        from_date = (datetime.now().date() - relativedelta(months=max_period)).strftime('%Y-%m-%d')


# 2. 한 종목씩 DB업데이트 호출
for idx, stock in self.stock_list.iterrows():
	sql = f"""
		SELECT *
		 FROM ST_PRICE_DAY
		WHERE market_cd = '{stock.market_cd}'
		  AND stock_cd = '{stock.stock_cd}'
		  AND stock_date >= '{from_date}'
			"""
	df_prices = pd.read_sql(sql, con=self.engine)
	for a in Analysis:
		if a.get_use_yn() == 'Y':
			batch_name = a.get_batch_name()
			analysis_period = a.get_timing_period()
			rs_prices = self.analysis_day_price(df_prices, batch_name, analysis_period)
			if rs_prices is None:
				continue
			print('[{}] #{:06d} {} ({}) : {} rows > ANALYSIS ST_PRICE_DAY [OK]'.format(datetime.now().strftime('%Y-%m-%d %H:%M'), idx + 1, stock['stock_cd'], stock['market_cd'], len(rs_prices)))
print('-------------FINISH UPDATE ST_PRICE_TIME-------------- ')

---

[적용 후기]

우선 코드가 한편 간결해졌고, 코드 내에서 공통 데이터를 관리함으로써 DB I/O도 줄일 수 있었다. 

실제 데이터가 저장되는 부분은 테이블에 직접 insert하고 조회하는 것이 맞지만, 공통코드 또는 마스터성 코드들은 이렇게 Enum 클래스를 이용해 코드에 녹여내는 것이 파이썬이 추구하는 객체 지향과 어느정도 맥락을 같이 하지 않나라는 생각을 해본다.

(실제 어느 글에서 말하길 배포 주기가 점점 짧아지면서, 히스토리 관리가 가능하고, 객체적으로 접근이 가능한 Enum클래스를 자주 사용한다는 글을 본 적이 있다.)  

 

 

프로젝트 경로)

github.com/sijune/timing-batch

sijune/timing-batch

 

참고)

Enum을 사용하는 이유

jojoldu.tistory.com/137

Enum 활용사례 3가지