간단한 모멘텀 전략 모의투자 (LS)#
KOSDAQ 지수의 이동평균선과 개별 종목의 모멘텀을 활용하는 전략을 구현하고, LS(구 이베스트투자)증권 모의투자 환경에서 검증해보도록 하겠습니다.
전략 만들기#
임의의 이름을 가진 디렉토리(폴더)를 만듭니다. 이 예제에서는 example이라는 디렉토리를 만들어 사용하겠습니다.
$ mkdir example
$ cd example
이제 파이썬 가상환경을 만듭니다.
$ python -m venv .venv
$ source .venv/bin/activate
Note
윈도우에서는 .venv\Scripts\activate 또는 .venv\Scripts\activate.bat 실행
전략 실행에 필요한 패키지는 requirements.txt 파일에 나열합니다.
pyqqq>=0.12.117
이후, 패키지를 설치합니다.
$ pip install -r requirements.txt
환경변수#
모의투자 계좌의 API 정보를 .env 파일에 저장합니다.
EBEST_APP_KEY=<your-app-key>
EBEST_APP_SECRET=<your-app-secret>
실행 설정#
app.yaml 파일을 생성하여 실행 설정을 지정합니다. 이번 전략 구현은 Hook 을 활용하도록 하겠습니다.
executor: hook
전략 코드#
간단한 모멘텀 전략을 구현한 샘플 코드입니다. 샘플 코드는 simple_momentum.py로 저장합니다.
"""
KOSDAQ 추세 전략
매수조건
- 전일 KOSDAQ 종가가 KOSDAQ 이동평균 3일,5일,10일 중 하나라도 높다면 진입
- 거래대금 기준 상위 5% 종목들 중 4일간의 종가 변동률 상위 20개 종목에 대해 매수
매도조건
- 보유종목 중 1일 이상 보유한 종목들은 시작가에 매도
- 장 중 3% 이상 수익이 발생한 종목은 익절
"""
import datetime as dtm
import pandas as pd
from decimal import Decimal
from typing import List
from pyqqq.datatypes import StockOrder, StockPosition
from pyqqq.brokerage.ebest.simple import EBestSimpleDomesticStock
from pyqqq.data.daily import get_all_ohlcv_for_date
from pyqqq.data.index import get_ohlcv_by_indices_for_period
from pyqqq.utils.market_schedule import (
get_last_trading_day,
get_trading_day_with_offset,
)
from pyqqq.utils.logger import get_logger
from pyqqq.utils.kvstore import KVStore
logger = get_logger("simple_momentum")
kvstore = KVStore("simple_momentum")
reserved_cash = 0.05 # 5% reserved cash
rate_per_asset = 0.05 # 5% per asset
def on_initialize() -> dict:
return {
"paper_trading": True,
"open_market_time_margin": dtm.timedelta(minutes=-10),
"interval": dtm.timedelta(seconds=60),
}
def on_market_open(
account_info: dict,
pending_orders: List[StockOrder],
positions: List[StockPosition],
broker_api: EBestSimpleDomesticStock,
):
sell_instructions = []
# 1일 이상 보유종목(전일 장중 익절하지 않은 종목)들을 모두 시작가에 매도한다.
for p in positions:
sell_instructions.append((p.asset_code, 0, -p.quantity))
buy_instructions = []
# KOSDAQ 지수가 상승추세가 아니라면 진입하지 않는다.
end_date = get_last_trading_day()
if check_market_trend_upward(get_trading_day_with_offset(end_date, offset_days=-9), end_date):
rank_df = get_ohlcv_with_value_rank_threshold(end_date, 95) # 거래대금 기준 상위 5% 종목들
price_change_df = get_price_changes(get_trading_day_with_offset(end_date, offset_days=-4), end_date) # 4일간의 종가 변동률 순 정렬된 종목들
common_index = rank_df.index.intersection(price_change_df.index)
target_df = price_change_df.loc[common_index] # 거래대금 상위 5% 종목들만 거래 대상으로 선정
for code, row in target_df.iterrows():
close_price = row["close"]
instruction = order_target_percent(account_info, positions, code, close_price, rate_per_asset)
if instruction[2] > 0:
buy_instructions.append(instruction)
if len(buy_instructions) >= 20: # 최대 20개 종목에 대해 매수 (5% * 20 = 100%)
break
return sell_instructions + buy_instructions
def trade_func(
account_info: dict,
pending_orders: List[StockOrder],
positions: List[StockPosition],
broker_api: EBestSimpleDomesticStock,
):
print_pending_orders(pending_orders)
print_positions(positions)
print_account_info(account_info)
instructions = []
pending_assets = set()
for o in pending_orders:
pending_assets.add(o.asset_code)
for p in positions:
# 3% 이상 수익이 발생한 종목은 익절한다.
if p.current_pnl >= 0.03:
if p.asset_code not in pending_assets:
instructions.append((p.asset_code, 0, -p.quantity))
return instructions
def handle_pending_orders(pending_orders, broker_api):
# 미체결 주문은 그대로 유지
return []
def check_market_trend_upward(start_date: dtm.date, end_date: dtm.date) -> bool:
"""
KOSDAQ 지수가 상승추세인지 확인합니다.
3일 이동평균선, 5일 이동평균선, 10일 이동평균선 중 하나라도 종가보다 높다면 상승추세로 판단합니다.
Args:
start_date: 시작일
end_date: 종료일
"""
kosdaq_df = get_kosdaq_ohlcv(start_date, end_date)
last_kosdaq_close = kosdaq_df["close"].iloc[-1]
kosdaq_ma3 = kosdaq_df["ma3"].iloc[-1]
kosdaq_ma5 = kosdaq_df["ma5"].iloc[-1]
kosdaq_ma10 = kosdaq_df["ma10"].iloc[-1]
a = last_kosdaq_close > kosdaq_ma3
b = last_kosdaq_close > kosdaq_ma5
c = last_kosdaq_close > kosdaq_ma10
return a or b or c
def get_kosdaq_ohlcv(start_date: dtm.date, end_date: dtm.date) -> pd.DataFrame:
"""KOSDAQ 지수의 OHLCV 데이터를 가져옵니다"""
# KOSDAQ 지수 코드 '2001'에 대한 데이터 조회
df = get_ohlcv_by_indices_for_period(["KOSDAQ"], start_date, end_date)
df = df["KOSDAQ"] # 단일 지수 데이터만 추출
# 이동평균 계산
df["ma3"] = df["close"].rolling(window=3).mean()
df["ma5"] = df["close"].rolling(window=5).mean()
df["ma10"] = df["close"].rolling(window=10).mean()
return df
def get_ohlcv_with_value_rank_threshold(target_date: dtm.date, threshold: int) -> pd.DataFrame:
"""특정 날짜의 OHLCV 데이터를 가져와서 value rank 기준으로 필터링합니다"""
df = get_all_ohlcv_for_date(target_date)
df["value_rank"] = df["value"].rank(ascending=True)
df["value_rank_ratio"] = df["value_rank"] / len(df) * 100
return df[df["value_rank_ratio"] > threshold].copy()
def get_price_changes(from_date: dtm.date, to_date: dtm.date) -> pd.DataFrame:
"""특정 기간 동안의 종가 변동률을 계산합니다"""
s_df = get_all_ohlcv_for_date(from_date)
e_df = get_all_ohlcv_for_date(to_date)
n_df = pd.DataFrame(index=s_df.index)
n_df["prev_close"] = s_df["close"]
n_df["close"] = e_df["close"]
n_df["pctchg"] = (n_df["close"] - n_df["prev_close"]) / n_df["prev_close"] * 100
n_df = n_df.sort_values(by="pctchg", ascending=False)
return n_df
def calc_quantity_by_percent(account_info, price, percent):
"""
주문량 계산
Args:
account_info: 계좌 정보. total_balance 필드가 있어야 함
price: 주문가격
percent: 목표 비중 (0.0 ~ 1.0)
Returns:
int: 주문량
"""
total_balance = account_info["total_balance"] * (1.0 - reserved_cash)
quantity = total_balance * percent / price
return int(quantity)
def order_target_percent(
account_info,
positions: List[StockPosition],
asset_code,
current_price,
target_percent,
):
"""목표 비중에 맞춰 매수/매도 주문 지시를 위한 tuple 생성"""
target_size = calc_quantity_by_percent(account_info, current_price, target_percent)
current_size = 0
for p in positions:
if p.asset_code == asset_code:
current_size = p.quantity
break
order_size = target_size - current_size
return (asset_code, current_price, order_size)
def print_pending_orders(pending_orders: List[StockOrder]):
logger.info("")
logger.info("=====================================")
logger.info("Pending Orders:")
logger.info("=====================================")
for p in pending_orders:
logger.info(f"{p.asset_code} {p.side} {p.order_type} {p.filled_quantity}/{p.quantity} @ {p.price}")
logger.info("=====================================")
def print_positions(positions: List[StockPosition]):
logger.info("")
logger.info("=====================================")
logger.info("Positions:")
logger.info("=====================================")
for p in positions:
logger.info(f"{p.asset_code} {p.quantity} @ {p.average_purchase_price} PnL: {p.current_pnl}")
logger.info("=====================================")
def print_account_info(account_info: dict):
logger.info(f"- Total balance: {account_info['total_balance']:,}원")
logger.info(f"- Purchase amount: {account_info['purchase_amount']:,}원")
logger.info(f"- Evaluated amount: {account_info['evaluated_amount']:,}원")
logger.info(f"- PnL rate: {account_info['pnl_rate'].quantize(Decimal('0.01'))}% ({account_info['pnl_amount']:,}원)")
이 전략의 주요 특징은 다음과 같습니다:
매수 조건#
KOSDAQ 지수의 종가가 3일, 5일, 10일 이동평균선 중 하나라도 높으면 매수 시그널
거래대금 기준 상위 5% 종목들 중에서 4일간의 종가 변동률이 높은 상위 20개 종목을 매수
매도 조건#
1일 이상 보유한 종목은 다음날 시작가에 매도
장중 3% 이상 수익 발생 시 익절
디렉토리 구조#
최종적으로 example 디렉토리 구성을 보면 다음과 같습니다.
.
├── .env
├── .venv
├── app.yaml
├── simple_momentum.py
└── requirements.txt
PC에서 실행하기#
전략 코드에 문제는 없는지 배포하기 이전에 PC에서 실행하여 봅니다. Hook 방식은 python 으로 실행이 되지 않고, qqq 명령줄 도구를 사용해야 합니다.
$ qqq run ./simple_momentum.py
실행 시 로그를 통해 주문 현황, 포지션, 계좌 정보 등을 확인할 수 있습니다.
배포하기#
qqq deploy 명령어를 사용해 배포할 수 있습니다.
위 명령어를 실행하면 디렉토리의 모든 파일을 압축하여 업로드 합니다.
$ qqq deploy ./simple_momentum.py
Deploying ./simple_momentum.py as simple_momentum
Uploading ./simple_momentum.py to GCS bucket
...