5분, 15분 이평선을 이용한 모의투자 (한투)#

5분과 15분 이동평균선을 활용해 골든크로스와 데드크로스 신호로 매매하는 전략을 구현하고, 이를 모의투자 환경에서 검증해보도록 하겠습니다. 모의투자는 한국투자증권에서 진행합니다.

전략 만들기#

임의의 이름을 가진 디렉토리(폴더)를 만듭니다. 이 예제에서는 example이라는 디렉토리를 만들어 사용하겠습니다.

$ mkdir example
$ cd example

이제 파이썬 가상환경을 만듭니다.

$ python -m venv .venv
$ source .venv/bin/active

Note

윈도우에서는 .venv\Scripts\activate 또는 .venv\Scripts\activate.bat 실행

전략 실행에 필요한 패키지는 requirements.txt 파일에 나열합니다.

pyqqq>=0.12.117

이후, 패키지를 설치합니다.

$ pip install -r requirements.txt

환경변수#

모의투자 계좌의 Open API 정보를 .env 파일에 저장합니다.

KIS_APP_KEY=<your-app-key>
KIS_APP_SECRET=<your-app-secret>
KIS_CANO=<your-account-number>
KIS_ACNT_PRDT_CD=<your-account-product-code>

전략 코드#

골든크로스, 데드크로스 전략을 구현한 샘플 코드입니다. 샘플 코드는 golden_dead_cross.py로 저장합니다.

from pyqqq.brokerage.kis.oauth import KISAuth
from pyqqq.brokerage.kis.simple import KISSimpleDomesticStock
from pyqqq.datatypes import OrderSide, OrderType
from pyqqq.utils.logger import get_logger
from pyqqq.utils.market_schedule import is_trading_time
import asyncio
import dotenv
import os
import pandas as pd

dotenv.load_dotenv()


class GoldenDeadCrossStrategy:
    """
    골든크로스, 데드크로스 전략을 구현한 클래스

    아래를 참고하여 한국투자증권의 API 정보를 환경변수 파일 (.env) 에 저장해야 함

    Examples:

        # 한국투자증권 API app key
        KIS_APP_KEY=<your-app-key>

        # 한국투자증권 API appsecret
        KIS_APP_SECRET=<your-app-secret>

        # 계좌번호 (계좌번호에서 '-' 앞부분. 12345678-01 이면 12345678)
        KIS_CANO=<your-account-no>

        # 계좌 상품 코드 (계좌번호에서 '-' 뒷부분. 12345678-01 이면 01)
        KIS_ACNT_PRDT_CD=<your-account-product-code>
    """

    logger = get_logger("GoldenDeadCrossStrategy")

    def __init__(self):
        app_key = os.getenv("KIS_APP_KEY")
        app_secret = os.getenv("KIS_APP_SECRET")
        account_no = os.getenv("KIS_CANO")
        account_product_code = os.getenv("KIS_ACNT_PRDT_CD")

        auth = KISAuth(app_key, app_secret, paper_trading=True)
        self.stock_api = KISSimpleDomesticStock(auth, account_no, account_product_code)

    async def run(self):
        while True:
            try:
                if is_trading_time():
                    await self.on_trade()
                else:
                    self.logger.info("market closed")

            except Exception as e:
                self.logger.exception(e)

            await asyncio.sleep(60)  # 1분마다 실행

    async def on_trade(self):
        """거래 로직"""
        self.logger.info("on_trade")

        asset_code = "233740"  # KODEX 코스닥150 레버리지
        df = self.get_minute_data_with_signals(asset_code)

        if not df.empty:
            # 신호가 발생하면 매수 또는 매도
            if df["golden_cross"].iloc[-1]:
                self.logger.info("golden cross signal")
                self.buy(asset_code)
            elif df["dead_cross"].iloc[-1]:
                self.logger.info("dead cross signal")
                self.sell(asset_code)
        else:
            # 개장 후 15분 이전까지는 이동평균선을 구할 수 없음.
            pass

    def get_minute_data_with_signals(self, asset_code: str) -> pd.DataFrame:
        """
        분봉 데이터와 골든크로스, 데드크로스 신호 계산

        Args:
            code (str): 종목코드

        Returns:
            pd.DataFrame: 분봉 데이터와 신호
        """
        df = self.stock_api.get_today_minute_data(asset_code)

        # 오름차순으로 정렬
        df.sort_index(ascending=True, inplace=True)

        # 5분 이동평균선과 15분 이동평균선 계산하고 골든크로스, 데드크로스 신호 확인
        df["ma5"] = df["close"].rolling(5).mean()
        df["ma15"] = df["close"].rolling(15).mean()
        df.dropna(inplace=True)

        df["golden_cross"] = (df["ma5"] > df["ma15"]) & (
            df["ma5"].shift(1) < df["ma15"].shift(1)
        )
        df["dead_cross"] = (df["ma5"] < df["ma15"]) & (
            df["ma5"].shift(1) > df["ma15"].shift(1)
        )
        return df

    def buy(self, asset_code: str):
        """ 매수 주문 """

        # 미체결 주문을 확인하여 매도 중이면 취소하고, 이미 매수 주문이 있으면 주문하지 않음.
        pending_orders = self.stock_api.get_pending_orders()
        for o in pending_orders:
            if o.asset_code == asset_code:
                if o.side == OrderSide.BUY:
                    self.logger.info("already ordered")
                    return
                else:
                    self.logger.info("cancel sell order")
                    self.stock_api.cancel_order(o.order_no)

        # 종목의 매수 가능 수량 조회
        data = self.stock_api.get_possible_quantity(asset_code, OrderType.MARKET)
        possible_quantity = data["quantity"]

        # 매수 가능 수량이 있으면 시장가 주문
        if possible_quantity > 0:
            self.stock_api.create_order(asset_code, OrderSide.BUY, possible_quantity, OrderType.MARKET)

    def sell(self, asset_code: str):
        """ 매도 주문 """

        pending_orders = self.stock_api.get_pending_orders()
        for o in pending_orders:
            if o.asset_code == asset_code:
                if o.side == OrderSide.SELL:
                    self.logger.info("already ordered")
                    return
                else:
                    self.logger.info("cancel buy order")
                    self.stock_api.cancel_order(o.order_no)

        positions = self.stock_api.get_positions()
        for p in positions:
            # 보유 종목 중에 asset_code에 해당하는 종목이 있으면 시장가에 매도
            if p.asset_code == asset_code and p.quantity > 0:
                self.stock_api.create_order(asset_code, OrderSide.SELL, p.quantity, OrderType.MARKET)


async def run():
    # 일반 전략 엔트리함수 구현
    await GoldenDeadCrossStrategy().run()


if __name__ == "__main__":
    # PC에서 실행할 땐 직접 호출
    asyncio.run(run())

  • KODEX 코스닥150 레버리지 ETF 가격의 5분, 15분 이동평균선을 활용합니다.

  • 배포시에는 run() 함수를 실행하게 됩니다.

  • PC에서는 직접 실행해서 테스트하면 됩니다. 대신 실행하는 부분은 __name__ == '__main__' 조건으로 감싸 줘야 배포 시 문제가 생기지 않습니다.

  • pandas의 데이터프레임을 이용하여 golden_cross, dead_cross를 손쉽게 찾을 수 있습니다.


>>> print(df)

                      open   high    low  close  volume        value  cum_volume     cum_value      ma5          ma15  golden_cross  dead_cross
time
2025-01-23 09:14:00  53600  53600  53500  53600   39237   2101891700     2951184  159783022700  53580.0  53660.000000         False       False
2025-01-23 09:15:00  53500  53700  53500  53700   95271   5107186600     3046455  164890209300  53600.0  53646.666667         False       False
...                    ...    ...    ...    ...     ...          ...         ...           ...      ...           ...           ...         ...
2025-01-23 15:36:00  53700  53700  53700  53700       0            0    13827190  744242339200  53700.0  53700.000000         False       False
2025-01-23 15:37:00  53700  53700  53700  53700  910152  48875162400    14737342  793117501600  53700.0  53700.000000         False       False

디렉토리 구조#

최종적으로 example 디렉토리 구성을 보면 다음과 같습니다.

.
├── .env
├── .venv
├── golden_dead_cross.py
└── requirements.txt

PC에서 실행하기#

전략 코드에 문제는 없는지 배포하기 이전에 PC에서 실행하여 봅니다.

$ python golden_dead_cross.py

프로그램 실행 시 초당 거래 건수를 초과했습니다. 라는 에러 메시지가 나타날 수 있지만, 이는 무시해도 됩니다. 모의투자 환경에서는 1초당 API 요청이 2회로 제한되어 있어 이로 인한 에러 메시지가 발생할 수 있습니다.

{"rt_cd":"1","msg_cd":"EGW00201","msg1":"초당 거래건수를 초과하였습니다."}
500 Server Error: Internal Server Error for url: https://openapivts.koreainvestment.com:29443/uapi/domestic-stock/v1/quotations/inquire-time-itemchartprice?FID_ETC_CLS_CODE=&FID_COND_MRKT_DIV_CODE=J&FID_INPUT_ISCD=233740&FID_INPUT_HOUR_1=132300&FID_PW_DATA_INCU_YN=N
Retrying in 0.5 seconds...

한국투자증권의 API 호출 유량

실전투자는 1초당 20건이고, 모의투자는 1초당 2건입니다.

배포하기#

qqq deploy 명령어를 사용해 배포할 수 있습니다. 위 명령어를 실행하면 디렉토리의 모든 파일을 압축하여 업로드 합니다.

$ qqq deploy ./golden_dead_cross.py
Deploying ./golden_dead_cross.py as golden_dead_cross
Uploading ./golden_dead_cross.py to GCS bucket
...
On this page
Show Source