코스피 상위 100위 120일선 자동매매

코스피 시가총액 상위 100종목 중 120일 이동평균선(MA120) '아래'에 있는 종목 스크리닝

사전 준비:
1. pip install pykrx pandas
동작 방식:
1. 기준일 코스피 시가총액 상위 100종목 조회

2. 각 종목의 최근 120거래일 종가로 120일 이동평균 계산

3. 기준일 종가가 MA120보다 낮은 종목만 필터링하여 출력
주의:
- pykrx는 KRX(한국거래소) 공개 데이터를 스크래핑하는 비공식 라이브러리라

가끔 응답이 느리거나 일시적으로 막힐 수 있습니다.

- 이 스크립트는 투자 자문이 아니라 기술적 조건에 따른 단순 필터링 도구입니다.

MA120 아래에 있다는 것 자체가 매수/매도 신호를 의미하지 않습니다.

(중요) https://data.krx.co.kr/ 로그인 후 아이디/패스워드를 환경변수로 설정해야함.
- 리눅스/맥 인경우

~/.zshrc나 ~/.bash_profile에 위 두 줄

export KRX_ID=""
export KRX_PW=""

- 윈도우 환경변수에 추가

# -*- coding: utf-8 -*-
"""
코스피 시가총액 상위 100종목 중 120일 이동평균선(MA120) '아래'에 있는 종목 스크리닝

사전 준비:
    pip install pykrx pandas

동작 방식:
    1. 기준일 코스피 시가총액 상위 100종목 조회
    2. 각 종목의 최근 120거래일 종가로 120일 이동평균 계산
    3. 기준일 종가가 MA120보다 낮은 종목만 필터링하여 출력

주의:
    - pykrx는 KRX(한국거래소) 공개 데이터를 스크래핑하는 비공식 라이브러리라
      가끔 응답이 느리거나 일시적으로 막힐 수 있습니다.
    - 이 스크립트는 투자 자문이 아니라 기술적 조건에 따른 단순 필터링 도구입니다.
      MA120 아래에 있다는 것 자체가 매수/매도 신호를 의미하지 않습니다.
"""

import time
from datetime import datetime, timedelta

import pandas as pd
from pykrx import stock


def get_trading_day(base_date: str) -> str:
    """base_date(YYYYMMDD)가 휴장일이면 가장 최근 거래일로 보정"""
    for i in range(10):
        d = (datetime.strptime(base_date, "%Y%m%d") - timedelta(days=i)).strftime("%Y%m%d")
        try:
            df = stock.get_market_ohlcv(d, d, "005930")  # 삼성전자로 개장 여부 확인
            if not df.empty:
                return d
        except Exception:
            continue
    raise RuntimeError("최근 거래일을 찾지 못했습니다.")


def get_top100_by_market_cap(base_date: str) -> pd.DataFrame:
    """기준일 코스피 시가총액 상위 100종목"""
    df = stock.get_market_cap_by_ticker(base_date, market="KOSPI")
    if df.empty or "시가총액" not in df.columns:
        raise RuntimeError(
            f"{base_date} 시가총액 데이터를 가져오지 못했습니다. "
            "pykrx 버전을 확인하거나(pip install -U pykrx) 다른 날짜로 시도해보세요."
        )
    df = df.sort_values("시가총액", ascending=False).head(100)
    df["종목명"] = [stock.get_market_ticker_name(t) for t in df.index]
    return df


def is_below_ma120(ticker: str, base_date: str):
    """해당 종목의 base_date 종가가 MA120보다 낮은지 계산"""
    start = (datetime.strptime(base_date, "%Y%m%d") - timedelta(days=250)).strftime("%Y%m%d")
    ohlcv = stock.get_market_ohlcv(start, base_date, ticker)
    if len(ohlcv) < 120:
        return None  # 상장 120거래일 미만 등 데이터 부족
    ma120 = ohlcv["종가"].rolling(window=120).mean().iloc[-1]
    close = ohlcv["종가"].iloc[-1]
    return close, ma120, close < ma120


def main():
    # 조회 기준일 (기본값: 어제 날짜, 필요시 직접 YYYYMMDD로 수정)
    yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
    base_date = get_trading_day(yesterday)
    print(f"[기준일] {base_date}\n")

    top100 = get_top100_by_market_cap(base_date)

    results = []
    for ticker, row in top100.iterrows():
        r = is_below_ma120(ticker, base_date)
        if r is None:
            continue
        close, ma120, below = r
        if below:
            results.append({
                "종목코드": ticker,
                "종목명": row["종목명"],
                "시가총액(억원)": round(row["시가총액"] / 1e8),
                "종가": close,
                "MA120": round(ma120, 1),
                "괴리율(%)": round((close - ma120) / ma120 * 100, 2),
            })
        time.sleep(0.2)  # 과도한 요청 방지

    result_df = pd.DataFrame(results).sort_values("괴리율(%)")

    print(f"시가총액 100위 내 MA120 아래 종목: 총 {len(result_df)}개\n")
    print(result_df.to_string(index=False))

    result_df.to_csv("kospi_top100_below_ma120.csv", index=False, encoding="utf-8-sig")
    print("\n결과를 kospi_top100_below_ma120.csv 로 저장했습니다.")


if __name__ == "__main__":
    main()

코스피 시가총액 상위 100종목 중 120일 이동평균선(MA120) '아래'에 있는 종목 스크리닝

Comments