OHLC data processing using Google Cloud

I try for some time to speed up processing for a list of tickers using Google Cloud Shell and Python code provided by Gemini and ChatGPT.So I have list of tickers like this:

TB,FMC,FNV,FOOD.L,FOX,FOXA,FRT,FSLR,FTI,FTNT,FTV,FXI,GBPAUD=X,GBPCAD=X,GBPNZD=X,GBPUSD=X,GD,GDXJ.DE,GE,GILD,GIS,GL,GLW,GM,GOLD,GOOG,GOOGL,GPC,GPN,GPS,GRMN,GS,GWW,HAL,HAS,HBAN,HBI,HCA,HD,HES,HIG,HII,HLT,HOG,HOLX,HON,HOOD,HPE,HPQ,HRB,HRL,HSIC,HST,HSY,HUM,HWM,IB01.L,IBCI.DE,IBKR,IBM,ICE,IDTL.L,IDXX,IEX,IFF,IFX.DE,IGLN.L,IGLO.L,IH2O.L,ILMN,INCY,INR.PA,INTC,INTU,IP,IPG,IPGP,IPRE.DE,IQV,IR,IRM,ISLN.L,ISRG,IT,ITA,ITW,IVZ,J,JBHT,JCI,JETS,JKHY,JNJ,JNPR,JPM,JWN,K,KEY,KEYS,KHC,KIM,KLAC,KMB,KMI,KMX,KO,KODK,KR,L,LDOS,LEA,LEG,LEN,LH,LHA.DE,LHX,LIN,LKQ,LLY,LMT,LNC,LNT,LOCK,LOW,LRCX,LUV,LVS,LW,LYB,LYV,M,MA,MAA,MAR,MAS,MBG.DE,MCD,MCHP,MCK,MCO,MDLZ,MDT,MET,METV,MMM,MRNA,MRO,MS,MSCI,MU,NCLH,NDAQ,NEE,NGAS.L,NICE,NIO,NVD.DE,NVO,NZDUSD=X,O,OD7F.DE,OPRA,OXY,PAH3.DE,PBR,PDD,PFE,PINS,PLD,PLTR,PRAR.DE,PSA,PSTG,PUM.DE,RACE,RBLX,RGLD,RIVN,RWE.DE,RY,RYAAY,SAP.DE,SCHB,SEA,SHOP,SHl.DE,SIE.DE,SMSN.L,SNY,SOL-USD,SONY,SOXX,SPG,SPOT,STM,TDOC,TGT,TLK,TLT,TM,TME,TSCO,TSLA,TSM,TTE,TTE.PA,TUR.PA,TWLO,TXN,UBER,UL,URA,USDCAD=X,USDILS=X,USDJPY=X,USDMXN=X,USDNOK=X,USDPLN=X,USDSEK=X,UST.PA,VIV,VLO,VOD,VOW3.DE,VTOP.DE,VX1.DE,WCLD.L,WHR,WU,X,XAD6.DE,XCS4.DE,XLE,XMLD.DE,XRP-USD,XRX,XX2D.DE,Z,ZM,ZPRP.DE,ZPRV.DE

This is the code that read data for this tickers( all code was writen by GPT, I have no coding skils ... just ideas :


import os
import yfinance as yf
import pandas as pd

def fetch_data(ticker):
# Fetch data for the last 5 years with weekly interval
weekly_data = yf.Ticker(ticker).history(period="5y", interval="1wk")

return weekly_data

def save_data(data, ticker, folder):
# Select only Date, Open, High, Low, Close columns
data = data[['Open', 'High', 'Low', 'Close']].reset_index()

# Format OHLC data to 2 decimals
data['Open'] = data['Open'].round(2)
data['High'] = data['High'].round(2)
data['Low'] = data['Low'].round(2)
data['Close'] = data['Close'].round(2)

filename = os.path.join(folder, f"{ticker}.csv")
data.to_csv(filename, index=False)

def update_data(tickers_file, data_folder):
# Load tickers from file
with open(tickers_file, 'r') as f:
tickers = f.read().strip().split(',')

# Check if data folder exists, create if not
if not os.path.exists(data_folder):
os.makedirs(data_folder)

tickers_with_errors = []

for ticker in tickers:
try:
# Fetch data for each ticker
data = fetch_data(ticker)

if len(data) > 0:
# Save data for each ticker
save_data(data, ticker, data_folder)

# Calculate the previous Friday
last_date = pd.to_datetime(data.index.max())
previous_friday = last_date - pd.DateOffset(days=last_date.weekday() - 4) # 4: Friday

# Print the correct date range
print(f"Data saved for {ticker} ")

# Delete last line if it's NOT a Friday, Saturday, or Sunday
filename = os.path.join(data_folder, f"{ticker}.csv")
df = pd.read_csv(filename)
last_date = pd.to_datetime(df.iloc[-1]['Date'])
if last_date.weekday() not in [4, 5, 6]: # 4: Friday, 5: Saturday, 6: Sunday
df = df[:-1] # Remove the last row
df.to_csv(filename, index=False)
print(f"Last line of {ticker}.csv deleted (week incomplete).")

else:
print(f"No data available for {ticker}.")

except Exception as e:
print(f"Error fetching data for {ticker}: {str(e)}")
tickers_with_errors.append(ticker)

return tickers_with_errors

if __name__ == "__main__":
tickers_file = 'tickers.txt'
data_folder = 'data'

tickers_with_errors = update_data(tickers_file, data_folder)

if tickers_with_errors:
print("\nTickers with errors:")
print(tickers_with_errors)

And this is the code that process OHLC data:


import pandas as pd
import matplotlib.pyplot as plt
import mplfinance as mpf
import argparse
import os
import datetime

def load_data(csv_file):
df = pd.read_csv(csv_file)
df['Date'] = pd.to_datetime(df['Date'], utc=True) # Ensure dates are parsed as UTC
df['Date'] = df['Date'].dt.tz_localize(None) # Remove timezone information
df.set_index('Date', inplace=True)
return df

def aggregate_data(df, timeframe):
resampling_rules = {
'W': 'W',
'2W': '2W',
'3W': '3W',
'5W': '5W',
'7W': '7W',
'9W': '9W',
'12W': '12W',
'15W': '15W',
'20W': '20W',
'25W': '25W',
'30W': '30W'
}

if timeframe not in resampling_rules:
raise ValueError(f"Unsupported timeframe: {timeframe}")

rule = resampling_rules[timeframe]
resampled_df = df.resample(rule).agg({
'Open': 'first',
'High': 'max',
'Low': 'min',
'Close': 'last'
}).dropna()

return resampled_df

def check_inside_bar(df, timeframe, ticker):
aggregated_df = aggregate_data(df, timeframe)
latest_bars = aggregated_df.iloc[-2:]

if len(latest_bars) < 2:
print(f"Not enough data to determine inside bar pattern for {ticker}.")
return None, None

latest_bar = latest_bars.iloc[-1]
previous_bar = latest_bars.iloc[-2]

# Check if latest bar is inside the previous bar (mother bar)
if (latest_bar['High'] <= previous_bar['High']) and (latest_bar['Low'] >= previous_bar['Low']):
# Check if the body of the latest bar is greater than the body of the previous bar
latest_bar_body = abs(latest_bar['Close'] - latest_bar['Open'])
previous_bar_body = abs(previous_bar['Close'] - previous_bar['Open'])

if latest_bar_body > previous_bar_body and latest_bar['Close'] > latest_bar['Open']:
latest_bar_end_date = latest_bar.name.to_pydatetime().date()
current_date = datetime.datetime.now().date()

print(f"PATERN found for {ticker}.")
print(f"Next Time Frame will end on: {latest_bar_end_date}")

if latest_bar_end_date > current_date:
print(f"Skipping {ticker} as the latest bar end date is in the future: {latest_bar_end_date}")
return None, None

return aggregated_df, latest_bar

return None, None

def save_data_to_text(df, filename):
with open(filename, 'w') as f:
for idx, row in df.iterrows():
line = f"{idx.date()}: {row['Open']:.2f}:{row['High']:.2f}:{row['Low']:.2f}:{row['Close']:.2f}\n"
f.write(line)

def plot_candlestick_chart(df, output_filename, data_filename):
df = df.iloc[-10:] # Select the last 10 bars for plotting
mpf.plot(df, type='candle', style='charles', title='Candlestick Chart', savefig=output_filename)
save_data_to_text(df, data_filename)
print(f"Candlestick chart saved as {output_filename}")
print(f"Data saved as {data_filename}")

def main():
parser = argparse.ArgumentParser(description='Process stock data.')
parser.add_argument('timeframe', type=str, help='Timeframe for aggregation (e.g., W, 2W, 3W, etc.)')
args = parser.parse_args()

data_folder = 'data'
output_folder = 'aggregated_data'
if not os.path.exists(output_folder):
os.makedirs(output_folder)

csv_files = [f for f in os.listdir(data_folder) if f.endswith('.csv')]

for csv_file in csv_files:
df = load_data(os.path.join(data_folder, csv_file))
ticker = csv_file.split('.')[0]
aggregated_df, latest_bar = check_inside_bar(df, args.timeframe, ticker)
if aggregated_df is not None:
timeframe_folder = os.path.join(output_folder, args.timeframe)
if not os.path.exists(timeframe_folder):
os.makedirs(timeframe_folder)
output_filename = os.path.join(timeframe_folder, f"{ticker}.png")
data_filename = os.path.join(timeframe_folder, f"{ticker}_data.txt")
plot_candlestick_chart(aggregated_df, output_filename, data_filename)

if __name__ == "__main__":
main()

For example for this run: sharpvinvest@cloudshell:~ (sharpinvest-428908)$ python insideBar.py W
PATERN found for EURAUD=X.
Next Time Frame will end on: 2024-07-07
Candlestick chart saved as aggregated_data/W/EURAUD=X.png
Data saved as aggregated_data/W/EURAUD=X_data.txt
PATERN found for NIO.
Next Time Frame will end on: 2024-07-14
Candlestick chart saved as aggregated_data/W/NIO.png
Data saved as aggregated_data/W/NIO_data.txt
PATERN found for BSX.
Next Time Frame will end on: 2024-07-14
Candlestick chart saved as aggregated_data/W/BSX.png
Data saved as aggregated_data/W/BSX_data.txt
PATERN found for SOL-USD.
Next Time Frame will end on: 2024-07-14
Candlestick chart saved as aggregated_data/W/SOL-USD.png
Data saved as aggregated_data/W/SOL-USD_data.txt
PATERN found for MU.
Next Time Frame will end on: 2024-07-14
Candlestick chart saved as aggregated_data/W/MU.png
Data saved as aggregated_data/W/MU_data.txt
sharpvinvest@cloudshell:~ (sharpinvest-428908)$

Not correctly calculate the end date of the cuostom time frame. Any attmepts to imrove the code using AI failed. I only got some sort of a result like this:

https://chatgpt.com/share/62d6f456-c...5-3a999e2acf27

My goal is to deploy the code as google cloud function that will execute according to sheddule process OHLC data to find a patern and then send the candlestick chart over email. But I.m still very far from my objective

New Files

"C:\Users\victor.rebeja\AppData\Local\Spotware\cTrader\abb70432efbee65d18af69e79fe8efe1\ctrader-cli.exe" run "C:\Users\victor.rebeja\Documents\cAlgo\Sources\Robots\D1.algo" --ctid=[email protected] --pwd-file="D:\password.pwd" --account=3706404 --symbol=EURUSD --period=D1 --full-access > "D:\log.txt" 2>&1


timeout /t 200 /nobreak
taskkill /IM "ctrader-cli.exe" /F
"C:\Users\victor.rebeja\AppData\Local\miniconda3\python.exe" "C:\Users\victor.rebeja\Desktop\uploadCSV.py"
"C:\Users\victor.rebeja\AppData\Local\miniconda3\python.exe" "C:\Users\victor.rebeja\Desktop\uploadLogs.py"
"C:\Users\victor.rebeja\AppData\Local\miniconda3\python.exe" "C:\Users\victor.rebeja\Desktop\uploadFlag.py"
echo. > "D:\log.txt"
echo. > "D:\logUpload.txt"
exit

uploadCSV


import os
from datetime import datetime
from google.cloud import storage
from google.oauth2 import service_account
# Set environment variable for Google Cloud credentials
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "D:/service_account.json"
# Google Cloud Storage setup
bucket_name = 'ctraderdata'
storage_client = storage.Client()
bucket = storage_client.bucket(bucket_name)
log_file_path = "D:/logUpload.txt"
def log_message(message):
"""Append a log message to the log file and print it to the console."""
with open(log_file_path, 'a') as log_file:
log_file.write(message + '\n')
print(message)
def upload_to_gcs(file_path, destination_blob_name):
"""Upload a file to Google Cloud Storage and log the process."""
try:
blob = bucket.blob(destination_blob_name)
blob.upload_from_filename(file_path)
message = f"{file_path} uploaded to {bucket_name} as {destination_blob_name} on {datetime.now().strftime('%d.%m.%Y %H:%M:%S')}"
log_message(message)
except Exception as e:
error_message = f"Failed to upload {file_path} to {bucket_name}: {e} on {datetime.now().strftime('%d.%m.%Y %H:%M:%S')}"
log_message(error_message)
def upload_all_csv_files(directory_path):
"""Upload all CSV files in the specified directory and log the actions."""
for root, _, files in os.walk(directory_path):
for file_name in files:
if file_name.lower().endswith('.csv'): # Check for CSV files
file_path = os.path.join(root, file_name)
# Use relative path for destination blob name
destination_blob_name = os.path.relpath(file_path, start=directory_path).replace("\", "/")
upload_to_gcs(file_path, destination_blob_name)
if __name__ == "__main__":
# Set the directory to watch for CSV files
directory_path = r"C:\ClickAlgo\logs"
upload_all_csv_files(directory_path)

uploadLOGS


import gspread
from google.oauth2 import service_account
from datetime import datetime
# Paths to the log files
log_file_path1 = "D:/log.txt"
log_file_path2 = "D:/logUpload.txt"
# Google Sheets setup
spreadsheet_id = '1LAVZNiYUitxNCWGI7pn_X9ZI4dZIy-jY2dOPFAMDPTk'
sheet_name1 = 'ctraderrun'
sheet_name2 = 'upload'
# Google service account credentials
credentials = service_account.Credentials.from_service_account_file(
'D:/service_account.json',
scopes=['https://www.googleapis.com/auth/spreadsheets', 'https://www.googleapis.com/auth/drive']
)
def upload_to_google_sheets(file_path, sheet_name):
"""Uploads content from a text file to a specific Google Sheets tab."""
try:
# Authorize with Google Sheets API
gc = gspread.authorize(credentials)
sheet = gc.open_by_key(spreadsheet_id).worksheet(sheet_name)

# Clear the sheet tab before each upload
sheet.clear()
# Read file content and prepare data for batch update
with open(file_path, 'r') as file:
lines = file.readlines()
# Create a list of lists for batch upload (each line becomes a row)
data = [[line.strip()] for line in lines]
# Update the sheet with the file's content, starting from the first row
sheet.insert_rows(data, 1) # Insert data starting from the first row
print(f"{file_path} successfully uploaded to Google Sheets tab '{sheet_name}' at {datetime.now().strftime('%d.%m.%Y %H:%M:%S')}")

except Exception as e:
print(f"Failed to upload {file_path} to Google Sheets tab '{sheet_name}': {e}")
if __name__ == "__main__":
# Upload each file to its respective tab in the Google Sheets document
upload_to_google_sheets(log_file_path1, sheet_name1) # Upload to 'ctraderrun' tab
upload_to_google_sheets(log_file_path2, sheet_name2) # Upload to 'upload' tab



uploadFLAG


import os
from datetime import datetime
from google.cloud import storage
# Set environment variable for Google Cloud credentials
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "D:/service_account.json"
# Google Cloud Storage setup
bucket_name = 'ctraderdata'
storage_client = storage.Client()
bucket = storage_client.bucket(bucket_name)
# Paths for the file to upload and the log file
file_path = "D:/FLAG.flag"
log_file_path = "D:/logUpload.txt"
def log_message(message):
"""Append a log message to the log file and print it to the console."""
with open(log_file_path, 'a') as log_file:
log_file.write(message + '\n')
print(message)
def upload_to_gcs(file_path, destination_blob_name):
"""Upload the specified file to Google Cloud Storage and log the process."""
try:
blob = bucket.blob(destination_blob_name)
blob.upload_from_filename(file_path)
message = f"{file_path} uploaded to {bucket_name} as {destination_blob_name} on {datetime.now().strftime('%d.%m.%Y %H:%M:%S')}"
log_message(message)
except Exception as e:
error_message = f"Failed to upload {file_path} to {bucket_name}: {e} on {datetime.now().strftime('%d.%m.%Y %H:%M:%S')}"
log_message(error_message)
if __name__ == "__main__":
# Define the blob name as the filename for simpler tracking
destination_blob_name = os.path.basename(file_path)
upload_to_gcs(file_path, destination_blob_name)

alpha vantage:

import requests
import pandas as pd
from datetime import datetime, timedelta
# Alpha Vantage API key
api_key = ""
# Define the tickers and the function name for daily data
tickers = ["EURUSD", "GBPUSD"]
function = "FX_DAILY"
# Calculate yesterday's date
yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
# Loop over each ticker and fetch data
for ticker in tickers:
# Define the API URL
url = f"https://www.alphavantage.co/query?function={function}&from_symbol={ticker[:3]}&to_symbol={ticker[3:]}&apikey={api_key}"

# Fetch the data from Alpha Vantage
response = requests.get(url)
data = response.json()

# Check if the data is in the expected format
if "Time Series FX (Daily)" in data:
# Extract the daily time series
time_series = data["Time Series FX (Daily)"]

# Check if yesterday's data is available
if yesterday in time_series:
# Extract yesterday's data
daily_data = time_series[yesterday]
daily_data['Date'] = yesterday

# Convert data to DataFrame
df = pd.DataFrame([daily_data])

# Rename columns to match the format: Date, Open, High, Low, Close
df = df.rename(columns={
"1. open": "Open",
"2. high": "High",
"3. low": "Low",
"4. close": "Close"
})

# Reorder columns to have Date first
df = df[['Date', 'Open', 'High', 'Low', 'Close']]

# Save the data to CSV (you can upload it to the bucket later)
csv_filename = f"{ticker}_D1.csv"
df.to_csv(csv_filename, index=False)
print(f"Saved {csv_filename} for {yesterday}")
else:
print(f"Data for {yesterday} is not available for {ticker}")
else:
print(f"Failed to retrieve data for {ticker}")