코인 선물 ai 자동매매 프로그램 만들기 2강 소스 코드

 

 

 

 

 

 

 

1. 바이비트에서 도지코인 perpetual 타임프레임 5분, 15분, 2시간의 볼린저밴드, rsi, 이동평균선 7, 50, 200선 과거 데이터 15개씩 가져 오는 파이썬 코드 입니다.  긁어서 "1.자료가져오기.py" 에  복사 붙여 넣기 하세요.

 

도지코인 말고 본인이 원하는 코인이 있다면 symbol 값을 수정해서 사용하면 됩니다. symbol을 수정했을 때는 해당 코인의 가격 소수점도 따라서 바꿔야합니다.

15개 말고 캔들 데이터를 더 가져 오고 싶으면 코드를 쳇지피티나 제미나이, 그록에게 코드를 보여주고 늘려달라고 하세요~

다른 보조지표를 사용하고 싶으면 그것도 코드를 붙여넣고 수정해달라고 하면 됩니다~

 

import ccxt
import pandas as pd
import numpy as np
import time
import os

# 설정
symbol = 'DOGEUSDT'
소수점 = 5
timeframes = ['5m', '15m', '2h']
ma_periods = [7, 50, 200]

# 현재 py 파일 경로
current_dir = os.path.dirname(os.path.abspath(__file__))
csv_path = os.path.join(current_dir, '도지ohlcv.csv')

# Bybit 객체 초기화
bybit = ccxt.bybit({
    'enableRateLimit': True,
})

# 가격 데이터를 가져오는 함수
def get_timeframe_close(symbol, timeframe, total_limit=220):
    current_time_ms = int(time.time() * 1000)

    if timeframe[-1] == 'm':
        minutes_back = total_limit * int(timeframe[:-1])
    elif timeframe[-1] == 'h':
        minutes_back = total_limit * 60 * int(timeframe[:-1])
    elif timeframe[-1] == 'd':
        minutes_back = total_limit * 1440
    elif timeframe[-1] == 'w':
        minutes_back = total_limit * 10080
    elif timeframe[-1] == 'M':
        minutes_back = total_limit * 43200
    since = current_time_ms - (minutes_back * 60 * 1000)

    ohlcv = bybit.fetch_ohlcv(symbol, timeframe, since=since, limit=total_limit)
    df = pd.DataFrame(ohlcv, columns=['datetime', 'open', 'high', 'low', 'close', 'volume'])
    df['datetime'] = pd.to_datetime(df['datetime'], unit='ms') + pd.Timedelta(hours=9)
    df.set_index('datetime', inplace=True)
    return df

# RSI 계산
def calculate_wilder_rsi(df, period=14):
    delta = df['close'].diff()
    gain = delta.where(delta > 0, 0)
    loss = -delta.where(delta < 0, 0)

    avg_gain = gain.rolling(window=period, min_periods=period).mean()
    avg_loss = loss.rolling(window=period, min_periods=period).mean()

    for i in range(period, len(df)):
        current_gain = gain.iloc[i]
        current_loss = loss.iloc[i]
        avg_gain.iloc[i] = (avg_gain.iloc[i-1] * (period - 1) + current_gain) / period
        avg_loss.iloc[i] = (avg_loss.iloc[i-1] * (period - 1) + current_loss) / period

    rs = avg_gain / avg_loss
    rsi = 100 - (100 / (1 + rs))
    df['RSI'] = rsi.round(2)
    return df

# 이동평균선 계산
def calculate_moving_averages(df, periods):
    for period in periods:
        df[f'SMA_{period}'] = df['close'].rolling(window=period).mean().round(소수점)
    return df

# 볼린저 밴드 계산
def calculate_bollinger_bands(df, window=20, std_multiplier=2):
    df['BB_Mid'] = df['close'].rolling(window=window).mean()
    df['BB_Std'] = df['close'].rolling(window=window).std(ddof=0)
    df['BB_Up'] = df['BB_Mid'] + std_multiplier * df['BB_Std']
    df['BB_Down'] = df['BB_Mid'] - std_multiplier * df['BB_Std']
    df[['BB_Mid', 'BB_Up', 'BB_Down']] = df[['BB_Mid', 'BB_Up', 'BB_Down']].round(소수점)
    return df

# 데이터 수집 및 저장
def collect_and_save():
    combined_data = pd.DataFrame()
    for timeframe in timeframes:
        df = get_timeframe_close(symbol, timeframe)
        df = calculate_moving_averages(df, ma_periods)
        df = calculate_bollinger_bands(df)
        df = calculate_wilder_rsi(df)
        df['timeframe'] = timeframe
        selected = df[['open', 'high', 'low', 'close', 
                       'SMA_7', 'SMA_50', 'SMA_200', 
                       'RSI', 'BB_Mid', 'BB_Up', 'BB_Down', 
                       'timeframe']].copy()
        selected = selected.tail(15).reset_index()
        combined_data = pd.concat([combined_data, selected], ignore_index=True)

    combined_data.to_csv(csv_path, index=False, encoding='utf-8-sig')
    print(f"CSV 파일이 생성되었습니다: {csv_path}")

# 실행
if __name__ == "__main__":
    collect_and_save()

 

 

 

2. 제미나이에게 데이터를 주고 분석하여 포지션을 추천 받는 코드입니다. 제미나이 api 키를 발급받아 입력하세요. instruction 부분에 본인이 원하는 프롬프트를 넣고 답변을 받을 수 있습니다.

 

import google.generativeai as genai
import google.api_core.exceptions
import pandas as pd
import os
import ast
import re
import json
from datetime import datetime
import logging
import time

# gRPC 경고 메시지 제거
os.environ["GRPC_VERBOSITY"] = "ERROR"
os.environ["GRPC_TRACE"] = ""

# 현재 경로 설정
current_directory = os.path.dirname(os.path.abspath(__file__))
csv_input_path = os.path.join(current_directory, '도지ohlcv.csv')
csv_output_path = os.path.join(current_directory, '도지market_analysis_results.csv')

# 로그 설정
logging.basicConfig(
    filename=os.path.join(current_directory, 'error_log.log'),
    level=logging.ERROR,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

# Google Generative AI API 키 리스트
api_keys = [
    "발급받은 api키를 여기에 입력하세요"
]

# 모델 리스트
models = ["gemini-2.0-flash", "gemini-2.0-flash-exp"]

current_key_index = 0
current_model_index = 0
attempt = 0
max_attempts = len(api_keys) * len(models)

def analyze_market_data(market_data_json):
    global attempt, current_key_index, current_model_index
    original_response = "No response received"

    instruction = f'''
Analyze the market data for 5m, 15m, and 2h timeframes. Analyze the flow through the 7, 50, and 200 moving averages, RSI, and Bollinger Bands given for each time frame. Focus on candlestick patterns such as Bullish/Bearish Engulfing, Hammer, Pin Bar, Morning/Evening Star, Doji, and Harami. Summarize each timeframe based on the last 15 candles and the latest one. Identify trend and momentum. Decide one position: "Long", "Short", or "Hold".

Explain your decision using specific patterns and timeframes (e.g., "Bullish Engulfing on 5m at 0.37500"). Be concise and accurate.

Output only a plain JSON object without code blocks. Example:

{{ 
  'justification': '5m. Bullish Engulfing at 0.37500 after a downtrend. 15m shows Hammer around 0.37400. 2h confirms with Morning Star. All align bullish → Long.', 
  'position': 'Long' 
}}

Rules:

Long: All three timeframes show bullish patterns, and 2h confirms.  
Short: All show bearish patterns, and 2h confirms.  
Hold: If signals conflict, 2h is unclear, or only one or two charts show signals.

Examples:

Case 1 (Long):  
5m: Hammer  
15m: Bullish Engulfing  
2h: Morning Star → Long

Case 2 (Hold):  
5m: Bearish Engulfing  
15m: Doji  
2h: No pattern → Hold

Now analyze the following market data and respond in a single JSON object.

Market Data:

{market_data_json}

'''

    while attempt < max_attempts:
        try:
            genai.configure(api_key=api_keys[current_key_index])
            model_name = models[current_model_index]
            model = genai.GenerativeModel(model_name)

            response = model.generate_content(instruction)
            original_response = response.candidates[0].content.parts[0].text

            result_text_cleaned = re.sub(r"^\s*```(?:json|python|.*)?\s*", '', original_response,
                                         flags=re.IGNORECASE | re.MULTILINE)
            result_text_cleaned = re.sub(r"```\s*$", '', result_text_cleaned, flags=re.MULTILINE).strip()

            start = result_text_cleaned.find('{')
            end = result_text_cleaned.rfind('}')
            if start != -1 and end != -1 and end > start:
                result_text_cleaned = result_text_cleaned[start:end+1]

            analysis_result = ast.literal_eval(result_text_cleaned)
            gemini_position = analysis_result.get('position', 'Unknown')
            now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            print(f"제미나이 응답 완료! 추천 포지션은 {gemini_position} 입니다. (시간: {now})")

            return result_text_cleaned, original_response, model_name

        except google.api_core.exceptions.ResourceExhausted:
            current_key_index = (current_key_index + 1) % len(api_keys)
            if current_key_index == 0:
                current_model_index = (current_model_index + 1) % len(models)
            attempt += 1
            logging.warning(
                f"Resource exhausted. Switching to API key index {current_key_index}, model index {current_model_index}."
            )

        except google.api_core.exceptions.InternalServerError:
            logging.error("Internal Server Error, retrying...", exc_info=True)
            time.sleep(3)
            attempt += 1
            continue

        except Exception as e:
            logging.error(f"Unhandled Error: {str(e)}", exc_info=True)
            current_key_index = (current_key_index + 1) % len(api_keys)
            if current_key_index == 0:
                current_model_index = (current_model_index + 1) % len(models)
            attempt += 1
            time.sleep(1)
            continue

    raise Exception("Max retries reached, unable to complete the request.")

def validate_positions(csv_file, gemini_position, gemini_justification):
    try:
        data = pd.read_csv(csv_file)
        data_5m = data[data['timeframe'] == '5m'].sort_values('datetime')
        data_15m = data[data['timeframe'] == '15m'].sort_values('datetime')
        data_2h = data[data['timeframe'] == '2h'].sort_values('datetime')

        rsi_5m_last4 = data_5m['RSI'].tail(3)
        rsi_15m_last4 = data_15m['RSI'].tail(3)
        rsi_2h_last4 = data_2h['RSI'].tail(3)

        def is_overbought_oversold(rsi_values):
            return any(r <= 30 or r >= 70 for r in rsi_values)

        if (
            is_overbought_oversold(rsi_5m_last4) or
            is_overbought_oversold(rsi_15m_last4) or
            is_overbought_oversold(rsi_2h_last4)
        ):
            return "Hold", gemini_justification

        latest_5m = data_5m.iloc[-1]
        latest_15m = data_15m.iloc[-1]
        latest_2h = data_2h.iloc[-1]

        def check_cross(row):
            if row['SMA_7'] > row['SMA_50']:
                return "golden"
            elif row['SMA_7'] < row['SMA_50']:
                return "dead"
            else:
                return "neutral"

        cross_5m = check_cross(latest_5m)
        cross_15m = check_cross(latest_15m)
        cross_2h = check_cross(latest_2h)

        if cross_5m == cross_15m == cross_2h and cross_5m in ["golden", "dead"]:
            if cross_5m == "golden" and gemini_position.lower() == "long":
                return gemini_position, gemini_justification
            elif cross_5m == "dead" and gemini_position.lower() == "short":
                return gemini_position, gemini_justification
            else:
                return "Hold", gemini_justification
        else:
            return "Hold", gemini_justification

    except Exception as e:
        logging.error(f"Error in validate_positions: {str(e)}", exc_info=True)
        return "Error", str(e)

def save_to_csv(position="N/A", justification="N/A", gemini_position="N/A", model="N/A"):
    result_df = pd.DataFrame({
        'datetime': [datetime.now()],
        'position': [position],
        'justification': [justification],
        'gemini_position': [gemini_position],
        'model': [model]
    })

    result_df.to_csv(csv_output_path, mode='a', index=False, header=not os.path.exists(csv_output_path))

if __name__ == "__main__":
    try:
        data = pd.read_csv(csv_input_path)
        market_data_json = data.to_json(orient='records')

        result_text_cleaned, original_response, used_model = analyze_market_data(market_data_json)

        try:
            analysis_result = ast.literal_eval(result_text_cleaned)
        except (ValueError, SyntaxError):
            try:
                json_string = result_text_cleaned.replace("'", '"')
                analysis_result = json.loads(json_string)
            except json.JSONDecodeError as e:
                logging.error("JSON Parsing Failed", exc_info=True)
                save_to_csv("Error", str(e), "N/A", used_model)
                exit()

        gemini_position = analysis_result.get('position', 'Unknown')
        gemini_justification = analysis_result.get('justification', 'No justification provided')

        validated_position, validated_justification = validate_positions(
            csv_input_path,
            gemini_position,
            gemini_justification
        )

        save_to_csv(validated_position, validated_justification, gemini_position, used_model)

    except Exception as e:
        logging.error(f"Unhandled Error: {str(e)}", exc_info=True)
        save_to_csv("Error", str(e), "N/A", "N/A")

 

 

 

 

바이비트 20% 할인 수수료 코드

22566

바이비트 가입 링크
https://partner.bybit.com/b/22566

후원
도지: DTd9cscTyrNWBigDoqgJr31S9YJMqntQHq