main.py Saham Turun v3 (belum bisa)
import json
import pandas as pd
import requests
import time
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
# ------------------------------
# 1️⃣ Baca config.json
# ------------------------------
with open('config.json') as f:
config = json.load(f)
SPREADSHEET_ID = config['spreadsheet_id']
TELEGRAM_TOKEN = config['telegram_token']
CHAT_ID = config['telegram_chat_id']
# ------------------------------
# 2️⃣ Range spreadsheet
# Ganti sesuai nama sheet & range
# ------------------------------
RANGE_NAME = 'Sheet1!A:F' # Sesuaikan dengan header kamu
# ------------------------------
# 3️⃣ Scope Google Sheets API
# ------------------------------
SCOPES = ['https://www.googleapis.com/auth/spreadsheets.readonly']
# ------------------------------
# 4️⃣ Fungsi buat service Google Sheets
# ------------------------------
def get_sheet_service():
creds = None
try:
creds = Credentials.from_authorized_user_file('token.json', SCOPES)
except:
flow = InstalledAppFlow.from_client_secrets_file(
'credentials.json', SCOPES)
creds = flow.run_local_server(port=0)
with open('token.json', 'w') as token:
token.write(creds.to_json())
service = build('sheets', 'v4', credentials=creds)
return service
# ------------------------------
# 5️⃣ Ambil data dari spreadsheet
# ------------------------------
def get_sheet_data(service):
sheet = service.spreadsheets()
result = sheet.values().get(spreadsheetId=SPREADSHEET_ID,
range=RANGE_NAME).execute()
values = result.get('values', [])
if not values:
print('No data found.')
return None
header = values[0]
rows = values[1:]
df = pd.DataFrame(rows, columns=header)
return df
# ------------------------------
# 6️⃣ Kirim pesan Telegram
# ------------------------------
def send_telegram_message(message):
url = f'https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage'
payload = {'chat_id': CHAT_ID, 'text': message}
response = requests.post(url, json=payload)
print(f"Telegram status: {response.status_code} | {response.text}")
# ------------------------------
# 7️⃣ Cek harga & kirim alert
# ------------------------------
def check_prices(df):
for _, row in df.iterrows():
kode = row['Kode Saham']
emiten = row['Nama Emiten']
try:
harga = float(row['Harga Saat Ini'])
batas_turun = float(row['Batas Turun'])
batas_naik = float(row['Batas Naik'])
except:
print(f"Skipping row: {row}")
continue
print(f"Memeriksa {kode}: {harga} | Turun: {batas_turun} | Naik: {batas_naik}")
if harga < batas_turun:
msg = f"🚨 [TURUN] {kode} - {emiten}\nHarga {harga:.2f} di bawah batas turun {batas_turun}"
send_telegram_message(msg)
elif harga > batas_naik:
msg = f"🚀 [NAIK] {kode} - {emiten}\nHarga {harga:.2f} di atas batas naik {batas_naik}"
send_telegram_message(msg)
# ------------------------------
# 8️⃣ Main loop
# ------------------------------
if __name__ == '__main__':
service = get_sheet_service()
while True:
df = get_sheet_data(service)
if df is not None:
check_prices(df)
print("Menunggu 60 detik...\n")
time.sleep(60)
Comments
Post a Comment