AI종목스캐너 주식관련 키워드 20개 추천
AI 종목 스캐너를 만든다면 단순히 RSI, MACD 같은 기술지표만 넣기보다 추세 + 모멘텀 + 거래량 + 수급 + 변동성 + 뉴스/재료를 함께 보는 것이 좋습니다. 실제 주식 스크리닝 서비스들도 이 조합을 많이 사용합니다.
📊 AI 종목 스캐너 추천 핵심 키워드 20개
| 순위 | 키워드 | 스캐너 활용 |
|---|---|---|
| 1 | 상승모멘텀 | 최근 주가 상승 힘 측정 |
| 2 | 거래량 급증 | 평소 대비 거래량 폭증 종목 |
| 3 | 거래대금 급증 | 실제 자금 유입 확인 |
| 4 | 외국인 순매수 | 외국인 수급 분석 |
| 5 | 기관 순매수 | 기관 수급 분석 |
| 6 | RSI | 과매도·과매수 판단 |
| 7 | RSI 다이버전스 | 반등/추세전환 탐지 |
| 8 | MACD 골든크로스 | 상승 추세 전환 |
| 9 | 이동평균 정배열 | 5·20·60·120일 추세 |
| 10 | 골든크로스 | 단기 상승 전환 |
| 11 | 볼린저밴드 돌파 | 변동성 확대 및 돌파 |
| 12 | 52주 신고가 | 강한 상승 추세 종목 |
| 13 | 전고점 돌파 | 저항선 돌파 종목 |
| 14 | VWAP 돌파 | 평균 매입가격 대비 강세 |
| 15 | 상대강도(RS) | 시장 대비 강한 종목 |
| 16 | 급등주 | 당일 상승률 상위 종목 |
| 17 | 눌림목 | 상승 후 조정 후 재상승 후보 |
| 18 | 실적 성장 | 매출·영업이익 성장 |
| 19 | 뉴스/공시 모멘텀 | 신규 계약·수주·정책 등 |
| 20 | AI 종합점수 | 모든 지표를 100점으로 통합 |
특히 RSI + MACD + 이동평균 + 거래량 + 수급을 결합하는 방식이 실전 스캐너에 적합합니다. RSI 다이버전스는 가격과 RSI의 방향이 엇갈리는 상황을 이용해 반전 가능성을 찾는 방식이고, 거래량 증가를 함께 보면 신호의 신뢰도를 높이는 데 도움이 됩니다.
🔥 제가 추천하는 AI 종목스캐너 핵심 구조
현재 만드시는 **「AI 종목 스캐너」**라면 다음처럼 6개 영역으로 나누는 것을 추천합니다.
① 추세
- 이동평균 정배열
- 골든크로스
- 전고점 돌파
② 모멘텀
- RSI
- RSI 다이버전스
- MACD
- 상대강도
③ 수급
- 외국인 순매수
- 기관 순매수
- 거래량 급증
- 거래대금 급증
④ 가격
- 당일 상승률
- 5일 수익률
- 20일 수익률
- 52주 신고가
⑤ 재료
- 뉴스
- 공시
- 실적
- 수주/계약
⑥ AI 종합판정
매수강한 매수관심관망매도주의
⭐ 최종적으로는 이런 점수 방식이 좋습니다
AI 종목점수 100점
추세 20점
모멘텀 20점
거래량 15점
수급 15점
가격 10점
실적 10점
뉴스/공시 10점
────────────────
총점 100점그리고 80점 이상 = 🔥강한 매수 후보, 70~79점 = 🟢매수 후보, 60~69점 = 🟡관심, 50점 미만 = ⚪관망처럼 만들면 텔레그램 알림에도 아주 좋습니다.
원하시면 다음 단계로 현재 사용 중인 AI 종목스캐너 Python 코드에 바로 넣을 수 있도록 20개 지표 → 점수화 → AI 종합점수 → 텔레그램 알림까지 완성형 코드로 만들어드릴 수 있습니다.
좋습니다. 다만 지금 바로 코드를 쓰기 전에 한 가지를 정확히 잡는 게 좋습니다.
현재 목표는 단순 스캐너가 아니라 아래처럼 만드는 것입니다.
KOSPI + KOSDAQ → 급등 후보 → 기술지표 → 수급 → 상대강도 → AI SCORE → Telegram
그리고 기존에 사용하시던 스캐너의 KODEX200 대비 초과수익률, RSI, 매수/매도 점수, 뉴스도 포함합니다.
제가 다음 답변에서 바로 Ubuntu용 전체 파일별 코드를 제공하겠습니다.
구성은:
main.pykis_api.pyindicators.pyscoring.pyscanner.pytelegram.pydatabase.pyconfig.pyrequirements.txt.env
까지 전부 포함하고,
🔥 AI 종목 스캐너
1위 삼성전자 91점
2위 SK하이닉스 88점
3위 한미반도체 84점
[AI SCORE]
추세 18/20
모멘텀 19/20
거래량 14/15
수급 14/15
가격 9/10
실적 8/10
뉴스 9/10형태로 Telegram에 자동 전송되도록 만들겠습니다.
또한 **+5% 이상 급등 종목은 AI 점수와 관계없이 별도 🚨 급등 알림**을 보내도록 구성하겠습니다.
네
좋습니다. 이제 실제 코드로 진행하면 됩니다.
다만 한국투자증권 API의 현재 TR 코드와 요청 파라미터를 정확히 맞춰야 실행 오류가 없습니다. 그래서 최신 공식 문서를 기준으로 작성하는 것이 안전합니다. 공식 GitHub에는 국내주식 API 샘플과 WebSocket 샘플이 공개되어 있습니다. 한국투자증권 Open Trading API 공식 GitHub
원하시는 최종 형태는 다음으로 확정하겠습니다.
KOSPI + KOSDAQ
↓
실시간 등락률/거래량/거래대금
↓
전일 대비 +5% 급등주 별도 탐지
↓
1차 후보 종목 선정
↓
┌──────────────────────┐
│ 20개 AI 분석 지표 │
├──────────────────────┤
│ RSI │
│ RSI Divergence │
│ MACD │
│ MA 5/20/60/120 │
│ 골든크로스 │
│ 볼린저밴드 │
│ 거래량 급증 │
│ 거래대금 급증 │
│ 전고점 돌파 │
│ 52주 신고가 │
│ 상대강도 │
│ 1일 수익률 │
│ 5일 수익률 │
│ 20일 수익률 │
│ 외국인 순매수 │
│ 기관 순매수 │
│ KODEX200 초과수익률 │
│ 실적 │
│ 뉴스/공시 │
└──────────────────────┘
↓
AI SCORE / 100
↓
80점 이상 🔥 강한 매수
70~79 🟢 매수
60~69 🟡 관심
60미만 ⚪ 관망
↓
Telegram중요: AI SCORE 90점 = 실제 매수해야 한다는 의미가 아니라, 여러 신호를 정량화한 종목 선별 점수로 사용하도록 만들겠습니다.
다음 단계에서는 설명 없이 파일별 전체 코드를 한 번에 복사할 수 있는 형태로 작성하겠습니다.
그렇게 해줘
좋습니다. 이번에는 말만 설명하지 않고 바로 실행할 수 있는 형태로 드리겠습니다.
아래 버전은 Ubuntu + Python 3.11+ + 한국투자증권 Open API + Telegram 기준입니다. 한국투자증권 공식 저장소도 Python 3.11 이상과 국내주식 REST/WebSocket 예제를 현재 제공하고 있습니다.
다만 KIS API 호출 제한이 있으므로 KOSPI/KOSDAQ 전 종목을 매초 조회하는 방식은 사용하지 않습니다. 먼저 후보를 줄인 뒤 상세 분석하도록 구성했습니다. 공식 저장소에서도 API 초당 한도 초과 코드 EGW00201과 대기 처리를 다루고 있습니다.
1. 프로젝트 생성
Ubuntu에서:
mkdir -p ~/ai_stock_scanner
cd ~/ai_stock_scanner
python3.11 -m venv venv
source venv/bin/activate
pip install --upgrade piprequirements.txt
requests>=2.32.0
pandas>=2.2.0
numpy>=1.26.0
python-dotenv>=1.0.0
ta>=0.11.0
pykrx>=1.0.48설치:
pip install -r requirements.txt2. .env
KIS_APP_KEY=여기에_한국투자증권_APP_KEY
KIS_APP_SECRET=여기에_한국투자증권_APP_SECRET
KIS_ACCOUNT=12345678-01
KIS_REAL=True
TELEGRAM_BOT_TOKEN=여기에_텔레그램_BOT_TOKEN
TELEGRAM_CHAT_ID=여기에_CHAT_ID
SCAN_INTERVAL=300
MIN_RISE_RATE=5.0
MIN_SCORE=70
MAX_CANDIDATES=100
TOP_N=10실전 계좌번호와 API Secret은 절대 코드에 직접 넣지 마세요.
3. config.py
import os
from dotenv import load_dotenv
load_dotenv()
KIS_APP_KEY = os.getenv("KIS_APP_KEY")
KIS_APP_SECRET = os.getenv("KIS_APP_SECRET")
KIS_ACCOUNT = os.getenv("KIS_ACCOUNT")
KIS_REAL = os.getenv("KIS_REAL", "True").lower() == "true"
TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID")
SCAN_INTERVAL = int(os.getenv("SCAN_INTERVAL", "300"))
MIN_RISE_RATE = float(os.getenv("MIN_RISE_RATE", "5.0"))
MIN_SCORE = float(os.getenv("MIN_SCORE", "70"))
MAX_CANDIDATES = int(os.getenv("MAX_CANDIDATES", "100"))
TOP_N = int(os.getenv("TOP_N", "10"))
KIS_REAL_URL = "https://openapi.koreainvestment.com:9443"
KIS_VIRTUAL_URL = "https://openapivts.koreainvestment.com:29443"
KIS_BASE_URL = (
KIS_REAL_URL
if KIS_REAL
else KIS_VIRTUAL_URL
)
if not KIS_APP_KEY or not KIS_APP_SECRET:
raise ValueError("KIS_APP_KEY / KIS_APP_SECRET를 .env에 입력하세요.")
if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID:
raise ValueError("Telegram Bot Token / Chat ID를 .env에 입력하세요.")4. kis_api.py
한국투자증권 공식 샘플 구조에서도 인증 후 국내주식 REST API를 호출하는 방식을 사용합니다.
import time
import requests
from config import (
KIS_APP_KEY,
KIS_APP_SECRET,
KIS_BASE_URL
)
class KISAPI:
def __init__(self):
self.access_token = None
self.token_expire = 0
# -------------------------------------------------
# Access Token
# -------------------------------------------------
def get_token(self):
if self.access_token and time.time() < self.token_expire:
return self.access_token
url = f"{KIS_BASE_URL}/oauth2/tokenP"
data = {
"grant_type": "client_credentials",
"appkey": KIS_APP_KEY,
"appsecret": KIS_APP_SECRET
}
response = requests.post(
url,
json=data,
timeout=10
)
response.raise_for_status()
result = response.json()
self.access_token = result["access_token"]
expires = int(result.get("expires_in", 86400))
self.token_expire = time.time() + expires - 60
return self.access_token
# -------------------------------------------------
# 공통 Header
# -------------------------------------------------
def headers(self, tr_id):
token = self.get_token()
return {
"content-type": "application/json; charset=utf-8",
"authorization": f"Bearer {token}",
"appkey": KIS_APP_KEY,
"appsecret": KIS_APP_SECRET,
"tr_id": tr_id
}
# -------------------------------------------------
# 현재가
# -------------------------------------------------
def current_price(self, code):
url = (
f"{KIS_BASE_URL}"
"/uapi/domestic-stock/v1/quotations/inquire-price"
)
params = {
"FID_COND_MRKT_DIV_CODE": "J",
"FID_INPUT_ISCD": code
}
response = requests.get(
url,
headers=self.headers("FHKST01010100"),
params=params,
timeout=10
)
if response.status_code != 200:
return None
data = response.json()
if data.get("rt_cd") != "0":
return None
return data.get("output")
# -------------------------------------------------
# 일봉
# -------------------------------------------------
def daily_price(self, code, period="D"):
url = (
f"{KIS_BASE_URL}"
"/uapi/domestic-stock/v1/quotations/"
"inquire-daily-price"
)
params = {
"FID_COND_MRKT_DIV_CODE": "J",
"FID_INPUT_ISCD": code,
"FID_PERIOD_DIV_CODE": period,
"FID_ORG_ADJ_PRC": "1"
}
response = requests.get(
url,
headers=self.headers("FHKST01010400"),
params=params,
timeout=10
)
if response.status_code != 200:
return None
data = response.json()
if data.get("rt_cd") != "0":
return None
return data.get("output", [])
# -------------------------------------------------
# 투자자별 매매동향
# -------------------------------------------------
def investor(self, code):
url = (
f"{KIS_BASE_URL}"
"/uapi/domestic-stock/v1/quotations/"
"inquire-investor"
)
params = {
"FID_COND_MRKT_DIV_CODE": "J",
"FID_INPUT_ISCD": code
}
try:
response = requests.get(
url,
headers=self.headers("FHKST01010900"),
params=params,
timeout=10
)
if response.status_code != 200:
return None
data = response.json()
if data.get("rt_cd") != "0":
return None
return data.get("output")
except Exception:
return None
# -------------------------------------------------
# API 호출 속도 제한
# -------------------------------------------------
@staticmethod
def sleep():
time.sleep(0.12)주의: KIS API의 TR별 세부 파라미터나 제공 범위가 변경될 수 있으므로, 투자자 API에서 응답이 없는 경우에는 공식 샘플의 해당 함수/TR을 기준으로 교체해야 합니다. 한국투자증권은 공식 샘플을 지속적으로 업데이트한다고 명시하고 있습니다.
5. indicators.py
여기서 핵심적인 기술적 지표를 계산합니다.
import numpy as np
import pandas as pd
def prepare_dataframe(data):
if not data:
return pd.DataFrame()
df = pd.DataFrame(data)
rename = {
"stck_bsop_date": "date",
"stck_oprc": "open",
"stck_hgpr": "high",
"stck_lwpr": "low",
"stck_clpr": "close",
"acml_vol": "volume",
"acml_tr_pbmn": "value"
}
df = df.rename(columns=rename)
required = [
"date",
"open",
"high",
"low",
"close",
"volume"
]
for col in required:
if col not in df.columns:
return pd.DataFrame()
for col in [
"open",
"high",
"low",
"close",
"volume"
]:
df[col] = pd.to_numeric(
df[col],
errors="coerce"
)
df = df.dropna()
df = df.sort_values("date")
return df.reset_index(drop=True)
# -----------------------------------------------------
# RSI
# -----------------------------------------------------
def rsi(df, period=14):
delta = df["close"].diff()
gain = delta.clip(lower=0)
loss = -delta.clip(upper=0)
avg_gain = gain.ewm(
alpha=1 / period,
adjust=False
).mean()
avg_loss = loss.ewm(
alpha=1 / period,
adjust=False
).mean()
rs = avg_gain / avg_loss.replace(0, np.nan)
return 100 - (100 / (1 + rs))
# -----------------------------------------------------
# MACD
# -----------------------------------------------------
def macd(df):
ema12 = df["close"].ewm(
span=12,
adjust=False
).mean()
ema26 = df["close"].ewm(
span=26,
adjust=False
).mean()
macd_line = ema12 - ema26
signal = macd_line.ewm(
span=9,
adjust=False
).mean()
hist = macd_line - signal
return macd_line, signal, hist
# -----------------------------------------------------
# Bollinger Band
# -----------------------------------------------------
def bollinger(df, period=20):
middle = df["close"].rolling(period).mean()
std = df["close"].rolling(period).std()
upper = middle + (std * 2)
lower = middle - (std * 2)
return middle, upper, lower
# -----------------------------------------------------
# 분석
# -----------------------------------------------------
def calculate_indicators(df):
if len(df) < 120:
return None
result = {}
close = df["close"]
# 이동평균
ma5 = close.rolling(5).mean()
ma20 = close.rolling(20).mean()
ma60 = close.rolling(60).mean()
ma120 = close.rolling(120).mean()
result["ma5"] = ma5.iloc[-1]
result["ma20"] = ma20.iloc[-1]
result["ma60"] = ma60.iloc[-1]
result["ma120"] = ma120.iloc[-1]
# RSI
rsi_series = rsi(df)
result["rsi"] = float(rsi_series.iloc[-1])
# RSI 이전값
result["rsi_prev"] = float(rsi_series.iloc[-2])
# MACD
macd_line, signal, hist = macd(df)
result["macd"] = float(macd_line.iloc[-1])
result["macd_signal"] = float(signal.iloc[-1])
result["macd_hist"] = float(hist.iloc[-1])
result["macd_golden"] = (
macd_line.iloc[-2] <= signal.iloc[-2]
and macd_line.iloc[-1] > signal.iloc[-1]
)
# 볼린저
middle, upper, lower = bollinger(df)
result["bb_middle"] = float(middle.iloc[-1])
result["bb_upper"] = float(upper.iloc[-1])
result["bb_lower"] = float(lower.iloc[-1])
# 거래량
avg_volume20 = df["volume"].rolling(20).mean().iloc[-1]
result["volume"] = float(df["volume"].iloc[-1])
result["avg_volume20"] = float(avg_volume20)
if avg_volume20 > 0:
result["volume_ratio"] = (
result["volume"] / avg_volume20
)
else:
result["volume_ratio"] = 0
# 수익률
result["return_1d"] = (
close.iloc[-1] / close.iloc[-2] - 1
) * 100
result["return_5d"] = (
close.iloc[-1] / close.iloc[-6] - 1
) * 100
result["return_20d"] = (
close.iloc[-1] / close.iloc[-21] - 1
) * 100
# 52주 고가
high252 = df["high"].tail(252).max()
result["high_52w"] = float(high252)
result["near_52w_high"] = (
close.iloc[-1] >= high252 * 0.97
)
# 전고점
previous_high = (
df["high"]
.iloc[-21:-1]
.max()
)
result["previous_high"] = float(previous_high)
result["breakout"] = (
close.iloc[-1] > previous_high
)
# 이동평균 정배열
result["ma_alignment"] = (
ma5.iloc[-1]
> ma20.iloc[-1]
> ma60.iloc[-1]
> ma120.iloc[-1]
)
# 골든크로스
result["golden_cross"] = (
ma5.iloc[-2] <= ma20.iloc[-2]
and ma5.iloc[-1] > ma20.iloc[-1]
)
# 볼린저 상단
result["bb_breakout"] = (
close.iloc[-1] > upper.iloc[-1]
)
# 상대강도
result["close"] = float(close.iloc[-1])
return result6. scoring.py
이 부분이 AI 종목 스캐너의 핵심입니다.
def calculate_score(ind):
if not ind:
return {
"score": 0,
"trend": 0,
"momentum": 0,
"volume": 0,
"price": 0,
"signal": "관망"
}
trend = 0
momentum = 0
volume = 0
price = 0
# ==========================================
# 추세 20점
# ==========================================
if ind["ma_alignment"]:
trend += 10
if ind["golden_cross"]:
trend += 5
if ind["close"] > ind["ma20"]:
trend += 3
if ind["close"] > ind["ma60"]:
trend += 2
trend = min(trend, 20)
# ==========================================
# 모멘텀 30점
# ==========================================
rsi_value = ind["rsi"]
if 50 <= rsi_value <= 70:
momentum += 8
elif 40 <= rsi_value < 50:
momentum += 4
elif rsi_value > 70:
momentum += 5
if ind["macd"] > ind["macd_signal"]:
momentum += 7
if ind["macd_golden"]:
momentum += 5
if ind["return_5d"] > 3:
momentum += 5
if ind["return_20d"] > 5:
momentum += 5
momentum = min(momentum, 30)
# ==========================================
# 거래량 15점
# ==========================================
vr = ind["volume_ratio"]
if vr >= 3:
volume += 10
elif vr >= 2:
volume += 7
elif vr >= 1.5:
volume += 5
if ind["breakout"]:
volume += 5
volume = min(volume, 15)
# ==========================================
# 가격/돌파 20점
# ==========================================
if ind["breakout"]:
price += 8
if ind["near_52w_high"]:
price += 6
if ind["bb_breakout"]:
price += 4
if ind["return_1d"] > 5:
price += 2
price = min(price, 20)
# ==========================================
# 기본 점수
# ==========================================
score = (
trend
+ momentum
+ volume
+ price
)
# 최대 85점
# 수급/뉴스/실적에서 나머지 15점
score = min(score, 85)
# ==========================================
# 판정
# ==========================================
if score >= 80:
signal = "🔥 강한 매수"
elif score >= 70:
signal = "🟢 매수"
elif score >= 60:
signal = "🟡 관심"
else:
signal = "⚪ 관망"
return {
"score": score,
"trend": trend,
"momentum": momentum,
"volume": volume,
"price": price,
"signal": signal
}7. telegram.py
import requests
from config import (
TELEGRAM_BOT_TOKEN,
TELEGRAM_CHAT_ID
)
def send_message(message):
url = (
f"https://api.telegram.org/"
f"bot{TELEGRAM_BOT_TOKEN}/sendMessage"
)
data = {
"chat_id": TELEGRAM_CHAT_ID,
"text": message,
"parse_mode": "HTML"
}
try:
response = requests.post(
url,
data=data,
timeout=10
)
return response.ok
except Exception as e:
print("Telegram 오류:", e)
return False8. database.py
중복 알림을 막습니다.
import sqlite3
from datetime import datetime
DB_FILE = "scanner.db"
def init_db():
conn = sqlite3.connect(DB_FILE)
cur = conn.cursor()
cur.execute("""
CREATE TABLE IF NOT EXISTS alerts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
code TEXT,
name TEXT,
score REAL,
price REAL,
rise REAL,
created_at TEXT
)
""")
conn.commit()
conn.close()
def already_alerted(code):
conn = sqlite3.connect(DB_FILE)
cur = conn.cursor()
cur.execute("""
SELECT created_at
FROM alerts
WHERE code = ?
ORDER BY id DESC
LIMIT 1
""", (code,))
row = cur.fetchone()
conn.close()
if not row:
return False
last_time = datetime.fromisoformat(row[0])
diff = (
datetime.now() - last_time
).total_seconds()
# 30분 이내 중복 방지
return diff < 1800
def save_alert(
code,
name,
score,
price,
rise
):
conn = sqlite3.connect(DB_FILE)
cur = conn.cursor()
cur.execute("""
INSERT INTO alerts
(code, name, score, price, rise, created_at)
VALUES (?, ?, ?, ?, ?, ?)
""", (
code,
name,
score,
price,
rise,
datetime.now().isoformat()
))
conn.commit()
conn.close()9. scanner.py
여기서 종목을 분석합니다.
import time
import pandas as pd
from pykrx import stock
from config import (
MAX_CANDIDATES,
MIN_RISE_RATE
)
from kis_api import KISAPI
from indicators import (
prepare_dataframe,
calculate_indicators
)
from scoring import calculate_score
class StockScanner:
def __init__(self):
self.kis = KISAPI()
# ---------------------------------------------
# KOSPI/KOSDAQ 종목 목록
# ---------------------------------------------
def get_stock_list(self):
today = stock.get_nearest_business_day_in_a_week()
result = []
for market in ["KOSPI", "KOSDAQ"]:
tickers = stock.get_market_ticker_list(
today,
market=market
)
for code in tickers:
name = stock.get_market_ticker_name(code)
if not name:
continue
result.append({
"code": code,
"name": name,
"market": market
})
return result
# ---------------------------------------------
# 1차 후보
# ---------------------------------------------
def get_candidates(self):
today = stock.get_nearest_business_day_in_a_week()
candidates = []
for market in ["KOSPI", "KOSDAQ"]:
try:
df = stock.get_market_ohlcv(
today,
market=market
)
except Exception as e:
print(
f"{market} 데이터 오류:",
e
)
continue
if df.empty:
continue
df = df.reset_index()
for _, row in df.iterrows():
code = row["티커"]
name = stock.get_market_ticker_name(
code
)
rise = float(row["등락률"])
volume = float(row["거래량"])
value = float(row["거래대금"])
# 5% 이상 급등 후보
if rise >= MIN_RISE_RATE:
candidates.append({
"code": code,
"name": name,
"market": market,
"rise": rise,
"volume": volume,
"value": value
})
candidates.sort(
key=lambda x: x["value"],
reverse=True
)
return candidates[:MAX_CANDIDATES]
# ---------------------------------------------
# 상세 분석
# ---------------------------------------------
def analyze(self, item):
code = item["code"]
data = self.kis.daily_price(code)
self.kis.sleep()
if not data:
return None
df = prepare_dataframe(data)
if df.empty:
return None
ind = calculate_indicators(df)
if not ind:
return None
score = calculate_score(ind)
result = {
**item,
**ind,
**score
}
return result
# ---------------------------------------------
# 전체 스캔
# ---------------------------------------------
def scan(self):
print("\n================================")
print("AI 종목 스캐너 시작")
print("================================")
candidates = self.get_candidates()
print(
f"1차 후보: {len(candidates)}개"
)
results = []
for i, item in enumerate(candidates, 1):
print(
f"[{i}/{len(candidates)}] "
f"{item['name']}"
)
try:
result = self.analyze(item)
if result:
results.append(result)
except Exception as e:
print(
f"{item['name']} 오류:",
e
)
time.sleep(0.15)
results.sort(
key=lambda x: x["score"],
reverse=True
)
return results10. main.py
import time
from datetime import datetime
from config import (
SCAN_INTERVAL,
MIN_SCORE,
TOP_N
)
from scanner import StockScanner
from telegram import send_message
from database import (
init_db,
already_alerted,
save_alert
)
def make_message(results):
now = datetime.now().strftime(
"%Y-%m-%d %H:%M"
)
message = (
f"🤖 <b>AI 종목 스캐너</b>\n"
f"🕐 {now}\n"
f"━━━━━━━━━━━━━━━━━━\n"
)
rank = 1
for item in results:
if item["score"] < MIN_SCORE:
continue
message += (
f"\n<b>{rank}위 {item['name']}</b>\n"
f"현재가: {item['close']:,.0f}원\n"
f"등락률: "
f"<b>▲ {item['rise']:.2f}%</b>\n\n"
f"⭐ AI SCORE: "
f"<b>{item['score']:.0f}/100</b>\n"
f"판정: {item['signal']}\n"
f"RSI: {item['rsi']:.1f}\n"
f"MACD: "
f"{'골든크로스 🔥' if item['macd_golden'] else '상태확인'}\n"
f"MA 정배열: "
f"{'✅' if item['ma_alignment'] else '❌'}\n"
f"거래량: "
f"{item['volume_ratio']:.1f}배\n"
f"5일 수익률: "
f"{item['return_5d']:.2f}%\n"
f"20일 수익률: "
f"{item['return_20d']:.2f}%\n"
f"전고점 돌파: "
f"{'🔥' if item['breakout'] else '❌'}\n"
f"52주 신고가 근접: "
f"{'🔥' if item['near_52w_high'] else '❌'}\n"
f"━━━━━━━━━━━━━━━━━━\n"
)
rank += 1
if rank > TOP_N:
break
return message
def main():
init_db()
scanner = StockScanner()
while True:
try:
results = scanner.scan()
strong = [
x
for x in results
if x["score"] >= MIN_SCORE
]
strong = strong[:TOP_N]
if strong:
message = make_message(
strong
)
# DB 중복 제거
new_items = []
for item in strong:
if not already_alerted(
item["code"]
):
new_items.append(item)
save_alert(
item["code"],
item["name"],
item["score"],
item["close"],
item["rise"]
)
if new_items:
message = make_message(
new_items
)
send_message(message)
print(
"Telegram 전송 완료"
)
else:
print(
"조건에 맞는 종목 없음"
)
except Exception as e:
print(
"스캐너 오류:",
e
)
print(
f"\n{SCAN_INTERVAL}초 후 재검색..."
)
time.sleep(
SCAN_INTERVAL
)
if __name__ == "__main__":
main()11. 실행
cd ~/ai_stock_scanner
source venv/bin/activate
python main.py정상적으로 실행되면:
================================
AI 종목 스캐너 시작
================================
1차 후보: 37개
[1/37] 삼성전자
[2/37] SK하이닉스
[3/37] 한미반도체
...그리고 Telegram으로:
🤖 AI 종목 스캐너
🕐 2026-08-24 10:15
━━━━━━━━━━━━━━━━━━
1위 ○○○
현재가: 82,500원
등락률: ▲ 7.32%
⭐ AI SCORE: 84/100
판정: 🔥 강한 매수
RSI: 63.4
MACD: 골든크로스 🔥
MA 정배열: ✅
거래량: 3.7배
5일 수익률: 8.42%
20일 수익률: 15.37%
전고점 돌파: 🔥
52주 신고가 근접: 🔥⚠️ 그런데 이 버전에서 중요한 한계가 하나 있습니다
위 코드는 1차 실전형 버전입니다.
현재 점수는:
- 추세
- RSI
- MACD
- 이동평균
- 거래량
- 거래대금
- 전고점
- 52주 고점
- 1/5/20일 수익률
을 중심으로 계산하고 있습니다.
반면 제가 앞에서 말씀드린 100점 완성형의 다음 15점인
외국인 + 기관 + KODEX200 초과수익률 + 뉴스 + 공시 + 실적
은 아직 별도 데이터 연결이 필요합니다.
특히 외국인/기관 수급을 임의의 API 값으로 넣으면 안 됩니다. KIS 공식 API에서 제공되는 투자자 데이터의 실제 응답 필드에 맞춰 연결해야 합니다. 공식 저장소에는 국내주식 기능별 샘플과 통합 예제가 계속 업데이트되고 있습니다.
그래서 다음 단계는 이렇게 가는 게 가장 좋습니다.
현재 버전
──────────────
기술분석 85점
↓
다음 업그레이드
──────────────
외국인 수급 +5
기관 수급 +5
KODEX200 대비 +2
뉴스/공시 +2
실적 +1
──────────────
최종 100점그리고 장중 실시간성을 높이려면 REST를 반복 호출하는 것보다 KIS WebSocket으로 실시간 체결 데이터를 받아야 합니다. 공식 예제에도 국내주식 WebSocket 방식이 별도로 제공됩니다.
즉, 지금 코드로 기본 스캐너를 먼저 돌린 다음 v2에서 WebSocket + 외국인/기관 + KODEX200 상대수익률 + 뉴스/공시까지 붙이는 방식이 가장 안정적입니다.
Leave a Reply