주식투자를 하다보니 아무래도 각종 지표나 자료 이런것을 찾아볼 때가 많이 있습니다.
특히나 요즘 같이 주식시장이 상승했다 하락할때는 다양한 지표를 참고하는데요.
참고하는 지표들중에서 가장 유명한 지표인 Fear and Greed 지표 데이터를 가지고 와서 그걸 텔레그램으로 아침마다 전송해주는 Telegram bot을 만들어 보겠습니다.
※ Fear and Greed 지표란?
Fear and Greed 지표는 CNN에서 집계하는 주식시장에서의 투자자들의 심리에 관한 지표입니다.
크게 아래와 같이 구분이 되는데 시장이 공포에 빠질때는 Fear (공포) 구간이고 시장이 끝도 없이 상승하고 호재로만 가득할때는 Greed(탐욕) 구간에 돌입하는 경우 입니다.
점수 범위 시장 심리 의미
| 0 ~ 24 | 극단적 공포 (Extreme Fear) | 투자자들이 극도로 불안해 매도세가 강함 → 잠재적 매수 기회 |
| 25 ~ 49 | 공포 (Fear) | 투자자 심리가 불안함, 시장이 다소 하락세 |
| 50 ~ 74 | 탐욕 (Greed) | 투자자들이 매수 성향, 시장 상승세 |
| 75 ~ 100 | 극단적 탐욕 (Extreme Greed) | 투자자 과열, 과매수 상태 → 조정 가능성 높음 |
- 0~49: 공포(Fear) → 투자자들이 매도 위주, 시장이 저평가될 가능성이 있음
- 50~100: 탐욕(Greed) → 투자자들이 매수 위주, 시장이 과열될 가능성이 있음
위처럼 이러한 심리적 현상을 수치화 한것이 공포 탐욕지수인데 이 지표는 CNN에서 집계해서 발표하는데 미국시장이 끝난 이후(한국시간 아침 8시)으로 최신화 되어서 집계가 된다고 합니다.
사이트는 아래와 같습니다.
https://edition.cnn.com/markets/fear-and-greed
보통 Fear and Greed 지표를 가져올때 해당 페이지에 접속해 크롤링하여 해당 지표를 수집하는데 이 방법 대신에 편법(?) 과 같은 방법을 사용해서 데이터를 수집해보겠습니다.
※ 크롤링이 아닌 비공식 API 활용
해당 API는 CNN에서 공식적으로 제공하는 API가 아니라 CNN의 내부에서 사용하는 API입니다.
이걸 사용해서 한번 데이터를 호출해서 데이터를 어떤 식으로 반환하는지 살펴보겠습니다
1.데이터 수집 클래스 만들기
Fear and Greed 지표를 수집하는 클래스를 만들어 줍니다.
class FearGreedCrawler:
"""Fear and Greed 지표 크롤러 (CNN 주식 시장)"""
def __init__(self):
self.url = FEAR_GREED_INDEX_URL
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
}
2.비공식 api 호출 메서드 만들기
api는 아래의 api 주소를 호출해주도록 합니다. 해당 api는 비공식이기 때문에 문서가 따로 존재하지 않으며,
이 api는 엔드포인트가 언제 갑자기 변경될지 모릅니다.
https://production.dataviz.cnn.io/index/fearandgreed/graphdata
해당 메서드를 호출해보니 제가 원하는 데이터인 바로 직전일의 fear and greed의 데이터는 fear and greed의 명칭으로 전송되고 있었습니다.
그럼 해당 양식에 맞게 코드를 아래와 같이 짜주겠습니다.
def _get_from_api(self) -> Optional[Dict]:
"""API를 통한 데이터 가져오기"""
try:
# CNN Fear & Greed Index API 엔드포인트
response = requests.get(api_url, headers=self.headers, timeout=10)
response.raise_for_status()
data = response.json()
if data and 'fear_and_greed' in data:
fng_data = data['fear_and_greed']
score = fng_data.get('score', 0)
# 소수점 반올림하여 정수로 변환
value = int(round(score))
rating = fng_data.get('rating', '').title()
# rating을 표준 분류로 변환
classification_map = {
'Extreme Fear': 'Extreme Fear',
'Fear': 'Fear',
'Neutral': 'Neutral',
'Greed': 'Greed',
'Extreme Greed': 'Extreme Greed'
}
classification = classification_map.get(rating, self._classify_value(value))
timestamp = fng_data.get('timestamp', self._get_timestamp())
# ISO 형식의 타임스탬프를 변환
if timestamp and 'T' in timestamp:
from datetime import datetime
try:
dt = datetime.fromisoformat(timestamp.replace('+00:00', ''))
timestamp = dt.strftime("%Y-%m-%d %H:%M:%S")
except:
pass
return {
'value': value,
'classification': classification,
'timestamp': timestamp
}
except Exception as e:
print(f"API 오류: {e}")
return None
def _classify_value(self, value: int) -> str:
"""값에 따른 분류"""
if value <= 25:
return "Extreme Fear"
elif value <= 45:
return "Fear"
elif value <= 55:
return "Neutral"
elif value <= 75:
return "Greed"
else:
return "Extreme Greed"
def _get_timestamp(self) -> str:
"""현재 타임스탬프 반환"""
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
반응형
- _get_from_api : 비공식 api 를 호출하여 데이터를 정제하고 반환하는 메서드
- _classify_value : 공포탐욕 지수의 수치에 따라 구분을 해주는 메서드
- _get_timestamp : 현재 시간으로 시간을 반환해주는 메서드
호출확인
_get_from_api 를 호출하여 정상적으로 호출이 되는것을 확인 했고 fear_and_greed 명칭으로 json 데이터 타입으로 변환된것을 확인했습니다.
3.호출 확인 테스트용 스크립트 만들기
이제는 테스트 스크립트를 작성해서 진짜 작동을 하는지 확인을 할 차례입니다.
import sys
import io
from datacollector import FearGreedCollector
# Windows 콘솔 인코딩 문제 해결
if sys.platform == 'win32':
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
def test_get_from_api():
"""_get_from_api 메서드 테스트"""
print("=" * 50)
print("_get_from_api 메서드 테스트 시작")
print("=" * 50)
# FearGreedCollector 인스턴스 생성
collector = FearGreedCollector()
# _get_from_api 메서드 호출
print("\nAPI에서 데이터를 가져오는 중...")
result = collector._get_from_api()
if result:
print("\n✅ API 호출 성공!")
print(f"📊 Fear and Greed 값: {result['value']}/100")
print(f"📈 분류: {result['classification']}")
print(f"🕐 타임스탬프: {result['timestamp']}")
print("\n전체 데이터:")
print(result)
else:
print("\n❌ API 호출 실패 또는 데이터를 가져올 수 없습니다.")
print("\n" + "=" * 50)
print("테스트 완료")
print("=" * 50)
if __name__ == "__main__":
test_get_from_api()
- test_get_from_api : get_from_api 메서드 테스트용 api 입니다.
해당 메서드를 호출해보고 제대로 호출이 되고 있는지 테스트 해보겠습니다.
이처럼 위 코드를 사용해서 테스트 또한 정상적으로 완료된것을 확인 할 수 있습니다.
value값은 10.51에서 반올림을 해서 제대로 11로 반환되고 있고 11이니 _classify_value 메서드에서 제대로 구분해서 전달 되는것 까지 확인 했습니다.
이제 데이터를 가져오는 것까진 작업을 완료했으니 다음엔 텔레그램에 봇을 만들어서 자동으로 특정시간에 전송을 하는 부분까지 구축하도록 하겠습니다.
전체 코드
import requests
from typing import Dict, Optional
from config import FEAR_GREED_INDEX_URL
class FearGreedCollector:
"""Fear and Greed 지표 크롤러 (CNN 주식 시장)"""
def __init__(self):
self.url = FEAR_GREED_INDEX_URL
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
}
def get_fear_greed_index(self) -> Optional[Dict]:
"""
Fear and Greed 지표 수집 (CNN API 를 통해 수집)
Returns:
dict: {
'value': int, # 0-100 값
'classification': str, # Extreme Fear, Fear, Neutral, Greed, Extreme Greed
'timestamp': str
}
"""
# CNN API를 통해 수집
api_result = self._get_from_api()
if api_result:
return api_result
else:
return None
def _get_from_api(self) -> Optional[Dict]:
"""API를 통한 데이터 가져오기"""
try:
# CNN Fear & Greed Index API 엔드포인트
response = requests.get(api_url, headers=self.headers, timeout=10)
response.raise_for_status()
data = response.json()
if data and 'fear_and_greed' in data:
fng_data = data['fear_and_greed']
score = fng_data.get('score', 0)
# 소수점 반올림하여 정수로 변환
value = int(round(score))
rating = fng_data.get('rating', '').title()
# rating을 표준 분류로 변환
classification_map = {
'Extreme Fear': 'Extreme Fear',
'Fear': 'Fear',
'Neutral': 'Neutral',
'Greed': 'Greed',
'Extreme Greed': 'Extreme Greed'
}
classification = classification_map.get(rating, self._classify_value(value))
timestamp = fng_data.get('timestamp', self._get_timestamp())
# ISO 형식의 타임스탬프를 변환
if timestamp and 'T' in timestamp:
from datetime import datetime
try:
dt = datetime.fromisoformat(timestamp.replace('+00:00', ''))
timestamp = dt.strftime("%Y-%m-%d %H:%M:%S")
except:
pass
return {
'value': value,
'classification': classification,
'timestamp': timestamp
}
except Exception as e:
print(f"API 오류: {e}")
return None
def _classify_value(self, value: int) -> str:
"""값에 따른 분류"""
if value <= 25:
return "Extreme Fear"
elif value <= 45:
return "Fear"
elif value <= 55:
return "Neutral"
elif value <= 75:
return "Greed"
else:
return "Extreme Greed"
def _get_timestamp(self) -> str:
"""현재 타임스탬프 반환"""
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def format_message(self, data: Dict) -> str:
"""메시지 포맷팅"""
if not data:
return "❌ Fear and Greed 지표를 가져올 수 없습니다."
value = data['value']
classification = data['classification']
# 이모지 선택
emoji = self._get_emoji(value)
# 메시지 구성
message = f"""
{emoji} <b>주식 시장 Fear and Greed 지표</b>
📊 <b>현재 값:</b> {value}/100
📈 <b>분류:</b> {classification}
🕐 <b>업데이트:</b> {data['timestamp']}
<i>데이터 출처: CNN Money</i>
""".strip()
return message
def _get_emoji(self, value: int) -> str:
"""값에 따른 이모지 반환"""
if value <= 25:
return "😱" # Extreme Fear
elif value <= 45:
return "😨" # Fear
elif value <= 55:
return "😐" # Neutral
elif value <= 75:
return "😊" # Greed
else:
return "🚀" # Extreme Greed
import sys
import io
from datacollector import FearGreedCollector
# Windows 콘솔 인코딩 문제 해결
if sys.platform == 'win32':
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
def test_get_from_api():
"""_get_from_api 메서드 테스트"""
print("=" * 50)
print("_get_from_api 메서드 테스트 시작")
print("=" * 50)
# FearGreedCollector 인스턴스 생성
collector = FearGreedCollector()
# _get_from_api 메서드 호출
print("\nAPI에서 데이터를 가져오는 중...")
result = collector._get_from_api()
if result:
print("\n✅ API 호출 성공!")
print(f"📊 Fear and Greed 값: {result['value']}/100")
print(f"📈 분류: {result['classification']}")
print(f"🕐 타임스탬프: {result['timestamp']}")
print("\n전체 데이터:")
print(result)
else:
print("\n❌ API 호출 실패 또는 데이터를 가져올 수 없습니다.")
print("\n" + "=" * 50)
print("테스트 완료")
print("=" * 50)
if __name__ == "__main__":
test_get_from_api()
반응형