Python Teknik Analiz 2

Borsa analizinde teknik göstergeler, yatırım kararlarının vazgeçilmez araçlarıdır. Bu yazımızda, Python kullanarak borsa hisselerinin kapanış verilerinden Wave Trend Osilatörü (WT) hesaplayan ve bu verilerden al-sat sinyalleri üreten gelişmiş bir sistem paylaşacağım.

Bu sistem, pandas, numpy, openpyxl, colorama ve glob gibi popüler Python kütüphanelerini kullanır. Önceki çalışamalarımızda ki Excel dosyalarından kapanış verilerini okur, WT1 ve WT2 eğrilerini hesaplar, Dip ve tepe bölgelerinde al-sat sinyalleri üretir,  Sonuçları Excel’e kaydeder ve Başarılı/başarısız hisse senetleri için rapor oluşturur.

Wave Trend İndikatörü Nedir?
Wave Trend (WT) osilatörü, fiyatın momentumu hakkında bilgi sağlayan ve özellikle dip/tepe bölgelerinde verdiği sinyallerle ön plana çıkan bir teknik göstergedir. İki bileşenden oluşur:

WT1: Ana sinyal çizgisi

WT2: WT1’in kısa vadeli hareketli ortalaması (sinyal çizgisi)

Alım/satım sinyalleri bu iki çizginin kesişimleriyle belirlenir. Ek olarak, aşırı alım (OB) ve aşırı satım (OS) seviyeleri ile sinyallerin güvenilirliği artırılır.

Kodun Temel İşlevleri

1. Wave Trend Hesaplama
Kodun kalbinde, calculate_wave_trend fonksiyonu yer alıyor. Burada;
ESA (Exponential Smoothed Average) hesaplanır.
ESA’den sapmalar üzerinden CI (Composite Index) elde edilir.
CI’nın üstel ortalaması WT1, onun da 4 dönemlik ortalaması WT2 olarak alınır.

WT1 ve WT2 arasındaki kesişimlere göre sinyaller üretilir: 

Buy (AL): WT1, WT2’nin üzerine çıkar

Sell(SAT): WT1, WT2’nin altına iner

BuyAtBottom(DIPTE AL): Aşırı satım bölgesinde al sinyali

SellAtTop(TEPEDE SAT): Aşırı alım bölgesinde sat sinyali

2. Dosya Taraması ve İşleme
Tüm hisse senedi kapanış verileri “StokData/Kapanis/” klasöründen otomatik olarak taranır. Her bir dosya için:

Tarih sıralaması yapılır. WT hesaplanır. Son sinyal (AL, SAT, DIPTE_AL, TEPEDE_SAT) analiz edilir.

Sonuçlar “StokData/WaveTrend/” klasörüne kaydedilir.

3. Sinyal Özeti ve Raporlama
Tüm hisseler için elde edilen sinyaller, Tarih ve WT1/WT2 değerleriyle birlikte özetlenir.

“StokData/wave_trend_signals_YYYYMMDD_HHMM.xlsx” adıyla kaydedilir.

Başarılı ve başarısız işlenen hisselerin listesi oluşturulur.

Aşağıda paylaşıcağım Python Çalışmasını  X_04_BorsaPin_WaveTrend.py olarak kayıt edebilirsiniz.

Python Kodu

import pandas as pd
import numpy as np
import os
from datetime import datetime
from colorama import Fore, init
import glob
"""
Borsapin StokData/Kapanis/ klasöründeki hisse kapanış datalarından Wave Trend Sinyallerini hesaplar.
www.kursatsenturk.com

"""

init(autoreset=True)

class WaveTrendCalculator:
def __init__(self, n1=10, n2=21, ob2=53, os2=-53):
self.n1 = n1 # ESA periyodu
self.n2 = n2 # CI periyodu
self.ob2 = ob2 # Overbought seviyesi
self.os2 = os2 # Oversold seviyesi
self.successful_files = []
self.failed_files = []
self.signal_results = []

def calculate_wave_trend(self, df):
"""Wave Trend hesaplama fonksiyonu"""
try:
close = df['Kapanış']

# ESA (Exponential Smoothed Average)
esa = close.ewm(span=self.n1, adjust=False).mean()

# D (Deviation)
d = (close - esa).abs().ewm(span=self.n1, adjust=False).mean()

# CI (Currency Index)
ci = (close - esa) / (0.015 * d)

# WT1 ve WT2 hesaplama
wt1 = ci.ewm(span=self.n2, adjust=False).mean()
wt2 = wt1.rolling(window=4).mean()

# Sinyaller
buy = (wt1 > wt2) & (wt1.shift(1) <= wt2.shift(1))
buy_at_bottom = buy & (wt2 < self.os2)
sell = (wt2 > wt1) & (wt2.shift(1) <= wt1.shift(1))
sell_at_top = sell & (wt2 > self.ob2)

return wt1, wt2, buy, buy_at_bottom, sell, sell_at_top

except Exception as e:
print(f"{Fore.RED}❌ Wave Trend hesaplama hatası: {e}")
return None, None, None, None, None, None

@staticmethod
def get_last_signal(df, ticker_name):
"""Son sinyali bulma fonksiyonu"""
try:
# NaN değerleri temizle
clean_df = df.dropna(subset=['WT1', 'WT2'])

if clean_df.empty:
return None, None, None, None

# Son tarihten geriye doğru sinyal ara
for i in reversed(clean_df.index):
row = clean_df.loc[i]
date = row['Tarih']
wt1 = row['WT1']
wt2 = row['WT2']

if row['BuyAtBottom']:
return 'DIPTE_AL', date, wt1, wt2
elif row['SellAtTop']:
return 'TEPEDE_SAT', date, wt1, wt2
elif row['Buy']:
return 'AL', date, wt1, wt2
elif row['Sell']:
return 'SAT', date, wt1, wt2

return None, None, None, None

except Exception as e:
print(f"{Fore.RED}❌ {ticker_name} sinyal arama hatası: {e}")
return None, None, None, None

def process_single_file(self, file_path):
"""Tek dosya için Wave Trend hesaplama"""
# ticker_name'i en başta tanımla
ticker_name = "BILINMEYEN"

try:
# Dosya adından hisse adını alma
file_name = os.path.basename(file_path)
ticker_name = os.path.splitext(file_name)[0]

print(f"{Fore.YELLOW} Wave Trend hesaplanıyor: {ticker_name}...")

# Excel dosyasını okuma
df = pd.read_excel(file_path)

# Gerekli sütunların varlığını kontrol etme
required_columns = ['Tarih', 'Kapanış']
missing_columns = [col for col in required_columns if col not in df.columns]

if missing_columns:
raise ValueError(f"Eksik sütunlar: {missing_columns}")

# Veri kontrolü
if df.empty or df['Kapanış'].isna().all():
raise ValueError("Kapanış verisi boş veya geçersiz")

# Tarihe göre sıralama
df = df.sort_values('Tarih').reset_index(drop=True)

# Wave Trend hesaplamaları
print(f"{Fore.CYAN} ⚡ Wave Trend parametreleri: n1={self.n1}, n2={self.n2}, ob={self.ob2}, os={self.os2}")

wt1, wt2, buy, buy_at_bottom, sell, sell_at_top = self.calculate_wave_trend(df)

if wt1 is None:
raise ValueError("Wave Trend hesaplama başarısız")

# Yeni DataFrame oluşturma
result_df = pd.DataFrame()
result_df['Hisse_Adi'] = [ticker_name] * len(df)
result_df['Tarih'] = df['Tarih']
result_df['Kapanış'] = df['Kapanış']
result_df['WT1'] = wt1.round(4)
result_df['WT2'] = wt2.round(4)
result_df['Buy'] = buy
result_df['BuyAtBottom'] = buy_at_bottom
result_df['Sell'] = sell
result_df['SellAtTop'] = sell_at_top

# Son sinyali bulma
signal_type, signal_date, last_wt1, last_wt2 = self.get_last_signal(result_df, ticker_name)

# Sinyal sonucunu kaydetme
if signal_type:
self.signal_results.append({
'Hisse_Adi': ticker_name,
'Sinyal': signal_type,
'Sinyal_Tarihi': pd.to_datetime(signal_date).date() if signal_date else None,
'WT1': round(last_wt1, 2) if last_wt1 else None,
'WT2': round(last_wt2, 2) if last_wt2 else None
})

# Renkli sinyal yazdırma
renk = Fore.GREEN if "AL" in signal_type else Fore.RED
print(renk + f" Son Sinyal: {signal_type} @ {signal_date}")
else:
print(f"{Fore.WHITE} ▫️ Aktif sinyal yok")

# Çıktı klasörünü oluşturma
output_folder = "StokData/WaveTrend/"
os.makedirs(output_folder, exist_ok=True)

# Dosya yolu
output_file = os.path.join(output_folder, f"{ticker_name}.xlsx")

# Excel'e kaydetme
result_df.to_excel(output_file, index=False)

print(f"{Fore.GREEN}✅ {ticker_name} Wave Trend verileri başarıyla kaydedildi.")
print(f"{Fore.BLUE} Konum: {output_file}")
print(f"{Fore.BLUE} Toplam satır: {len(result_df)}")

self.successful_files.append(ticker_name)
return True

except Exception as e:
print(f"{Fore.RED}❌ {ticker_name} için hata: {e}")
self.failed_files.append(ticker_name)
return False

@staticmethod
def find_input_files(input_folder="StokData/Kapanis/"):
"""Giriş dosyalarını bulma"""
try:
# Excel dosyalarını arama
pattern = os.path.join(input_folder, "*.xlsx")
files = glob.glob(pattern)

if not files:
print(f"{Fore.RED}❌ {input_folder} klasöründe Excel dosyası bulunamadı!")
return []

print(f"{Fore.BLUE} {len(files)} adet Excel dosyası bulundu.")
return files

except Exception as e:
print(f"{Fore.RED}❌ Dosya arama hatası: {e}")
return []

def save_signals_summary(self, filename=None):
"""Sinyal özetini kaydetme"""
if not self.signal_results:
print(f"{Fore.YELLOW}⚠️ Kaydedilecek sinyal yok.")
return

if filename is None:
timestamp = datetime.now().strftime('%Y%m%d_%H%M')
filename = f"StokData/wave_trend_signals_{timestamp}.xlsx"

try:
# Klasör oluşturma
os.makedirs(os.path.dirname(filename), exist_ok=True)

# Sinyal DataFrame'i oluşturma
signals_df = pd.DataFrame(self.signal_results)

# Sinyal türüne göre sıralama
signal_order = ['DIPTE_AL', 'AL', 'SAT', 'TEPEDE_SAT']
signals_df['sinyal_order'] = signals_df['Sinyal'].map({s: i for i, s in enumerate(signal_order)})
signals_df = signals_df.sort_values(['sinyal_order', 'Sinyal_Tarihi'], ascending=[True, False])
signals_df = signals_df.drop('sinyal_order', axis=1)

# Excel'e kaydetme
signals_df.to_excel(filename, index=False)

print(f"{Fore.GREEN} Sinyal özeti kaydedildi: {filename}")
print(f"{Fore.BLUE} Toplam sinyal: {len(signals_df)}")

# Sinyal türü dağılımı
signal_counts = signals_df['Sinyal'].value_counts()
for signal, count in signal_counts.items():
color = Fore.GREEN if "AL" in signal else Fore.RED
print(f"{color} {signal}: {count} adet")

except Exception as e:
print(f"{Fore.RED}❌ Sinyal özeti kaydetme hatası: {e}")

def print_summary(self):
"""Özet rapor"""
total = len(self.successful_files) + len(self.failed_files)
success_rate = (len(self.successful_files) / total * 100) if total > 0 else 0

print(f"\n{Fore.CYAN} ===== WAVE TREND HESAPLAMA RAPORU =====")
print(f"{Fore.BLUE} Parametreler: n1={self.n1}, n2={self.n2}, OB={self.ob2}, OS={self.os2}")
print(f"{Fore.GREEN}✅ Başarılı: {len(self.successful_files)}")
print(f"{Fore.RED}❌ Başarısız: {len(self.failed_files)}")
print(f"{Fore.BLUE} Başarı oranı: {success_rate:.1f}%")
print(f"{Fore.MAGENTA} Toplam sinyal: {len(self.signal_results)}")

if self.successful_files:
print(f"{Fore.GREEN} Başarılı dosyalar: {', '.join(self.successful_files[:10])}")
if len(self.successful_files) > 10:
print(f"{Fore.GREEN} ... ve {len(self.successful_files) - 10} dosya daha")

if self.failed_files:
print(f"{Fore.RED} Başarısız dosyalar: {', '.join(self.failed_files)}")

def save_failed_list(self, filename="basarisiz_wavetrend_dosyalari.txt"):
"""Başarısız dosyaları kaydetme"""
if self.failed_files:
try:
with open(filename, 'w', encoding='utf-8') as f:
f.write("# Başarısız Wave Trend hesaplama dosyaları\n")
f.write(f"# Tarih: {datetime.now().strftime('%Y-%m-%d')}\n")
f.write(f"# Parametreler: n1={self.n1}, n2={self.n2}, OB={self.ob2}, OS={self.os2}\n\n")
for file_name in self.failed_files:
f.write(f"{file_name}\n")
print(f"{Fore.YELLOW} Başarısız dosyalar {filename} dosyasına kaydedildi.")
except Exception as e:
print(f"{Fore.RED}❌ Başarısız dosya listesi kaydetme hatası: {e}")

def main(self):
"""Ana fonksiyon"""
print(f"{Fore.CYAN} Wave Trend Hesaplama Sistemi Başlatılıyor...")
print(f"{Fore.BLUE} Parametreler: n1={self.n1}, n2={self.n2}, OB={self.ob2}, OS={self.os2}")

# Giriş dosyalarını bulma
input_files = self.find_input_files()

if not input_files:
return

print(f"{Fore.BLUE} Toplam işlenecek dosya: {len(input_files)}\n")

# Dosyaları işleme
for i, file_path in enumerate(input_files, 1):
print(f"\n{Fore.MAGENTA}[{i}/{len(input_files)}] İşleniyor...")
self.process_single_file(file_path)

# Sonuçları kaydetme ve raporlama
self.save_signals_summary()
self.save_failed_list()
self.print_summary()

print(f"\n{Fore.GREEN} Wave Trend hesaplama işlemi tamamlandı!")
print(f"{Fore.BLUE} Detay dosyaları: StokData/WaveTrend/")
print(f"{Fore.BLUE} Sinyal özeti: StokData/Wave_Trend_Sinyal_*.xlsx")

# Kullanım
if __name__ == "__main__":
# Özelleştirilebilir parametreler
calculator = WaveTrendCalculator(
n1=10, # ESA periyodu
n2=21, # CI periyodu
ob2=53, # Overbought seviyesi
os2=-53 # Oversold seviyesi
)

calculator.main()

Bu Bist Tarama projesininin 6. makalesi oldu sanırım. Bir sonraki çalışmamızda yukarıda paylaştığım python betiğinin (wavetrend osilatör) TradingView üzerinde çalışan indikatör kodlarını paylaşacağım, sonrasında Python Bist Tarama çalışmaları için Osilatör değerleri yada Linear Regresyon kanalları ile ilgili betikler paylaşabilirim.