연속하락 종목찾기

목표설정 전 다양한 방법 구상

본 페이지에서는 전종목 데이터를 본격적으로 활용할 방법에 대해 알아볼것입니다. 데이터를 모았으면 목적에 맞는 분석이 필요합니다. 어떤것을 해보실껀가요? 다양한 방법들이 있습니다.

연속 하락/상승 종목 찾기: 일정 기간 동안 지속적으로 하락하거나 상승하는 종목을 식별하여 추세를 분석할 수 있습니다.

거래량 급증 종목 분석: 평소보다 거래량이 급격히 증가한 종목을 찾아 시장의 관심도를 파악할 수 있습니다.

변동성 분석: 고가와 저가의 차이를 통해 종목의 변동성을 측정하고, 리스크를 평가할 수 있습니다.

이동평균선 분석: 단기, 중기, 장기 이동평균선을 활용하여 추세의 방향과 강도를 파악할 수 있습니다.

상관관계 분석: 종목간의 가격 움직임 상관관계를 분석하여 포트폴리오 구성에 활용할 수 있습니다.

계절성 분석: 특정 시기에 반복되는 주가 패턴을 찾아 시장의 주기성을 파악할 수 있습니다.

딥러닝을 활용한 예측: 다음과 같은 딥러닝 모델들을 활용할 수 있습니다

•

LSTM(Long Short-Term Memory): 시계열 데이터 예측에 특화된 모델

•

Prophet: Meta(구 Facebook)에서 개발한 시계열 예측 라이브러리로, 간단한 코드로 구현 가능

•

CNN(Convolutional Neural Network): 차트 패턴을 이미지로 변환하여 분석

투자전략에는 이외에도 많은 방법이 있습니다. 데이터를 수집하는 목적과 활용방안을 다각도로 고려하여 체계적인 분석을 수행해야 합니다. 반드시 투자에 국한되는것은 아닙니다.

예를 들어, 이러한 데이터 분석 기법들은 주식 시장 외에도 다양한 분야에서 활용될 수 있습니다. 날씨 패턴 분석, 소비자 행동 예측, 에너지 사용량 추적, 또는 스포츠 경기 결과 예측 등 시계열 데이터가 존재하는 모든 영역에서 응용이 가능합니다.

목표설정

이번에는 개별 주가 데이터에서 Change를 추출하여 최근 연속적으로 하락하는 종목을 찾아보겠습니다. 하락 종목을 찾은 후, 하락 원인을 분석하여 건전한 기업임에도 외국인 투자자들의 이탈과 같은 단기적 이슈로 인해 하락한 경우를 식별하고자 합니다.

이러한 경우, 향후 상승 가능성이 있어 매수 기회가 될 수 있습니다. 우리의 접근 방식은 다음과 같이 확장될 수 있습니다

•

재무분석 연계: 연속 하락종목의 재무제표를 수집하여 기업 건전성을 판단

•

지수 연계분석: 코스피/코스닥 지수 대비 상대강도가 우수했던 종목 중 일시적 하락 종목 포착

데이터 호출용 함수

데이터 읽기 페이지의 기존 함수를 개선하여 Open, High, Low, Close, Volume, Change 데이터를 하나로 병합하는 함수를 만들겠습니다. 이렇게 하면 분석에 필요한 데이터를 매개변수로 쉽게 호출할 수 있습니다.

import pyarrow.dataset as ds
import pandas as pd
import time
from concurrent.futures import ThreadPoolExecutor

def getTrading(column='Close'):
    '''column종류:Open, High, Low, Close, Volume, Change''' 
    dataFolder = "data"
    dataset = ds.dataset(dataFolder, format="parquet")
    def processFragment(fragment):
        code = fragment.path.split('/')[-1].split('.')[0]  # 파일명에서 종목 코드 추출
        table = fragment.to_table(columns=['Date', column])  # 필요한 컬럼만 읽기
        table = table.rename_columns(['Date', code])  # Close 컬럼을 종목 코드로 변경
        df = table.to_pandas().set_index('Date')
        return df
    startTime = time.time() 
    with ThreadPoolExecutor() as executor:
        dfs = list(executor.map(processFragment, dataset.get_fragments()))
    result = pd.concat(dfs, axis=1)
    endTime = time.time() 
    print(f"데이터 처리 실행 시간: {endTime - startTime:.4f}초")
    return result

df = getTrading('Change')
df
Python
복사

dataFolder 변수가 "data"로 지정된 이유는 초기 폴더명을 "data"로 설정했기 때문입니다. 만약 다른 폴더명을 사용하고 계시다면, 해당 폴더명으로 수정해주셔야 합니다. 더 나은 방법으로는 함수를 개선하여 폴더명을 매개변수로 받도록 수정하는 것입니다.

함수 상세 설명

읽기 페이지에서 코드를 설명했지만 함수로 변환한 코드를 다시 하나씩 자세히 살펴보겠습니다:

1. 필요한 라이브러리 임포트

•

pyarrow.dataset: 대용량 데이터를 효율적으로 처리하기 위한 라이브러리입니다.

•

pandas: 데이터 분석을 위한 기본 라이브러리입니다.

•

time: 함수의 실행 시간을 측정하기 위해 사용됩니다.

•

ThreadPoolExecutor: 여러 작업을 동시에 처리하여 속도를 향상시키는데 사용됩니다.

2. getTrading 함수 구조

이 함수는 주식 데이터를 읽어오는 메인 함수입니다. column 매개변수를 통해 원하는 데이터(시가, 종가 등)를 선택할 수 있습니다.

3. 내부 함수 processFragment

•

파일명에서 종목코드 추출: 각 파일의 이름에서 종목코드를 가져옵니다.

•

필요한 컬럼만 선택: 날짜와 사용자가 지정한 데이터 컬럼만 읽어옵니다.

•

컬럼명 변경: 데이터 컬럼의 이름을 종목코드로 바꿔줍니다.

•

인덱스 설정: 날짜를 인덱스로 지정합니다.

4. 병렬 처리 과정

ThreadPoolExecutor를 사용하여 여러 파일을 동시에 처리함으로써 실행 속도를 크게 향상시킵니다.

5. 결과 생성

•

데이터 결합: 모든 개별 데이터프레임을 하나로 합칩니다.

•

실행시간 측정: 함수의 실행 시간을 계산하여 출력합니다.

사용 예시

# 종가 데이터 가져오기
closeData = getTrading('Close')

# 거래량 데이터 가져오기
volumeData = getTrading('Volume')

# 등락률 데이터 가져오기
changeData = getTrading('Change')
Python
복사

결과

상장법인목록함수

데이터프레임을 종목코드로 생성했기 때문에 현재는 회사명을 확인할 수 없습니다. 연속 하락 종목을 찾은 후에는 상장법인목록을 활용하여 종목코드와 회사명을 매칭하여 출력하도록 하겠습니다.

import requests
def getCorpList():
    url = 'https://kind.krx.co.kr/corpgeneral/corpList.do'
    data = {
        'method': 'download',
        'orderMode': '1',
        'orderStat': 'D',
        'searchType': '13',
        'fiscalYearEnd': 'all',
        'location': 'all',
    }
    r = requests.post(url, data=data)
    df = pd.read_html(r.content, header=0)[0] 
    df['종목코드'] = df['종목코드'].astype(str)
    df['종목코드'] = df['종목코드'].str.zfill(6)
    return df
Python
복사

이 함수는 스크래핑의 기술 페이지에서 예시로 작성한 코드를 개선한 버전입니다. 자세한 내용은 링크를 참고해 주세요.

연속하락 종목 찾기

하락종목을 찾기 위해 데이터프레임의 데이터를 분석하는데 pandas를 적극 활용할 것입니다. pandas 라이브러리 페이지에서 다루지 않은 내용은 이 페이지에서 추가로 설명하고, 추후 정리하여 pandas 페이지에 업데이트하겠습니다.

7일 연속 하락종목

7일연속으로 하락하는 종목을 찾겠습니다. 각 종목의 일간 등락률 데이터를 확인하여 음수값(하락)을 True로 표시하는 새로운 데이터프레임을 생성합니다. 이렇게 만들어진 True/False 데이터프레임에 rolling 함수를 적용하여 7일 단위로 움직이는 윈도우를 생성하고, 각 윈도우 내에서 True값의 합계를 계산합니다. 7일 연속으로 하락한 경우 이 합계값이 7이 됩니다. 최근 데이터를 기준으로 이 합계값이 정확히 7인 종목들을 필터링하여 연속 하락 종목 리스트를 만듭니다.

마지막으로, 찾아낸 종목코드들을 이용하여 상장법인 목록에서 해당 종목들의 상세 정보를 추출합니다. 이를 통해 단순한 종목코드가 아닌, 회사명과 같은 추가 정보를 포함한 최종 결과를 얻을 수 있습니다. 이러한 분석은 투자자들이 시장의 하락 추세를 파악하고 투자 결정을 내리는 데 도움을 줄 수 있습니다.

관련지식 pandas : rolling, iloc, isin

import pandas as pd
# 상장법인 목록 가져오기
corpList = getCorpList()

# 연속 하락일수 설정 (7일)
minDeclineDays = 7  

# 일간 등락률이 음수인지 확인하여 True/False로 구성된 데이터프레임 생성
# df는 getTrading('Change')로 가져온 등락률 데이터
isDecline = df < 0

# 연속 하락 여부 계산
# rolling() 함수로 지정된 기간(minDeclineDays)동안의 윈도우를 만들고
# sum()으로 True값의 개수를 계산 (True=1, False=0)
# 예: 7일 연속 하락이면 sum()값이 7이 됨
rollingDecline = isDecline.rolling(window=minDeclineDays).sum()

# 최근 n일 동안 연속 하락한 종목 찾기
# iloc[-1]로 가장 최근 날짜의 데이터를 선택하고
# minDeclineDays와 같은 값을 가진 종목을 필터링
recentDecline = rollingDecline.iloc[-1] == minDeclineDays

# 결과 출력
# recentDecline[recentDecline]으로 True인 값만 선택하고
# index.tolist()로 해당 종목코드들의 리스트를 생성
decliningList = recentDecline[recentDecline].index.tolist()
print(f"{minDeclineDays}일 연속 하락한 종목 ({len(decliningList)})개:", decliningList)

# 상장법인 목록에서 연속 하락 종목들의 정보를 필터링하여 출력
# isin() 함수로 종목코드가 decliningList에 포함된 행만 선택
corpList[corpList['종목코드'].isin(decliningList)]
Python
복사

1) df

df에는 change데이터가 담겨 있습니다.

2) isDecline = df < 0

df의 Change 데이터가 0보다 작은 경우를 True/False로 표시하는 새로운 불리언 데이터프레임을 생성합니다. True는 1값을가지게 되고 False는 0값을 가지게 됩니다.

3) rolling

True와 False는 각각 1과 0의 값을 가지게 됩니다. 이때 rolling을 이용해서 7일 연속의 값을 합산합니다. minDeclineDays 변수에는 7일로 설정해뒀습니다.

rollingDecline = isDecline.rolling(window=minDeclineDays).sum()
Python
복사

4) 마지막 일자에 값이 7인경우

rolling 함수는 연속된 값들의 합을 계산합니다. True(1)와 False(0) 값에서 7거래일 연속 하락한 경우, True가 7번 연속 나타나 합계가 7이 됩니다. 이 계산된 값이 설정한 minDeclineDays(7)와 일치하는지 여부를 불리언 값으로 저장합니다.

recentDecline = rollingDecline.iloc[-1] == minDeclineDays  #7일
# rollingDecline.iloc[-1] 가장 마지막 행 즉, 2024-12-27일의 모든행을 minDeclineDays과 같으면 True아니면 False
Python
복사

5) True인것의 종목코드만 리스트로 가저오기

recentDecline[recentDecline]은 True값만 선택합니다. .index를 사용해 데이터프레임에서 종목코드만 추출하고, tolist로 이를 리스트 형태로 변환합니다.

decliningList = recentDecline[recentDecline].index.tolist()
Python
복사

6) 회사명으로 출력

isin() 함수는 데이터프레임에서 특정 값들이 포함되어 있는지 확인하는 기능을 합니다. corpList['종목코드'].isin(decliningList)는 상장법인 목록의 종목코드들 중에서 연속 하락한 종목 코드 리스트(decliningList)에 포함된 것들만 True로 표시합니다. 그리고 corpList[...]로 감싸서 True로 표시된 종목들의 정보만 최종적으로 추출합니다.

corpList[corpList['종목코드'].isin(decliningList)]
Python
복사