import hashlib as md5 import json import os import os.path import random from os import path from time import sleep import github import re from datetime import datetime """import discord import asyncio discordbot = discord.Client() async def send_discord_message(msg): await discordbot.wait_until_ready() channel = discordbot.get_channel(id=899353294244315250) # replace with channel_id await channel.send(msg) @discordbot.event async def on_ready(): #print('Logged in as') #print(discordbot.user.name) #print(discordbot.user.id) #print('------') # discordbot.start('OTQyODYzNzAwMTAzMTM5Mzc5.YgqsPw.cGgb4-vCT8fto9GeHj9srK1AejA') # send_discord_message("hi from the bot!") # discordbot.loop.create_task(send_discord_message("hi from the bot ^.^")) """ def edit_strategy_content(code, strategyContent_original, class_name_modified): strategyContent_modified_text = strategyContent_original rows = strategyContent_original.splitlines() class_name_original = "" for row in rows: if "(IStrategy)" in row: split_by_word = row.split() for word in split_by_word: if "(IStrategy)" in word: split_by_bracket_on = word.split("(") class_name_original = split_by_bracket_on[0] strategyContent_modified_text = \ strategyContent_modified_text.replace( class_name_original, class_name_modified) strategyContent_modified_text = "# source: " + code.download_url + "\n" + strategyContent_modified_text return strategyContent_modified_text, class_name_original, class_name_modified class github_downloader: # discordbot.run_until_complete(send_discord_message("I'm a bot ... beep - boop!")) githubAuth = "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" g = github.Github(githubAuth, timeout=2000) # Keyword to search dir_namex = "github_downloader" dir_namex_original = dir_namex + "/original/" dir_namex_modified = dir_namex + "/modified/" dir_name = os.getcwd() + "/" + dir_namex dir_name_original = os.getcwd() + "/" + dir_namex_original dir_name_modified = os.getcwd() + "/" + dir_namex_modified if not os.path.exists(dir_name): os.mkdir(dir_name) if os.path.exists(dir_name): if not os.path.exists(dir_name_original): os.mkdir(dir_name_original) if not os.path.exists(dir_name_modified): os.mkdir(dir_name_modified) limit = float('inf') # limit = 100 keyword = " freqtrade strategy IStrategy " \ " NOT backtest_validation NOT edge_positioning NOT hyperopt_interface " \ " NOT plotting NOT IResolver " sleep_timer = 3 exception_sleep_timer = 60 skiponduplicate = False class_name_original = "" class_name_modified = "" strategyContent_original = "" strategyContent_modified_text = "" file_name_with_path_original = "" file_name_with_path_modified = "" class_name_original = "" class_name_modified = "" sha_original = "" sha_modified = "" def __init__(self): self.fnames = None self.unique = None self.duplicate = None self.cwd = None def main(self): self.sortvalue = 'indexed' self.ordervalue = 'asc' # #print("Sorting: " + self.sortvalue + " / " + "Order: " + self.ordervalue) file_names = dict() self.fnames = [] self.unique = [] self.cwd = os.getcwd() self.duplicate = False for filename in os.listdir(self.cwd + "/" + self.dir_namex_original): with open(self.cwd + "/" + self.dir_namex_original + "/" + filename, "rb") as f: bytes = f.read() filehash = md5.sha256(bytes).hexdigest() self.fnames.append(filename) if filehash not in self.unique: self.unique.append(filehash) else: # #print("Removing Duplicate File " + filename + " not unique") os.remove(self.cwd + "/" + self.dir_namex_original + "/" + filename) sizes = ['size:0..2000', 'size:2000..3000', 'size:3000..4000', 'size:4000..5000', 'size:5000..7000', 'size:7000..15000', 'size:15000..27000', 'size:27000..500000', 'size:500000..'] # sizes = list(range(0, 80000))[::500] for index, currentSize in enumerate(sizes): # fileSizeFilter = currentSize # fileSizeFilter = 'size:' + str(currentSize) + ".." + str(sizes[index+1]) # currentDate = str(year) + "-" + str(month).zfill(2) + "-01" # dateRangeFilter = "created:" + previousDate + ".." + currentDate for code_try in range(3): try: codes = self.g.search_code(self.keyword + " language:python " + currentSize, sort=self.sortvalue, order=self.ordervalue) break except Exception: sleep(self.exception_sleep_timer) # #print("error produced on codes generation, retry...") continue for code_try in range(3): try: if codes.totalCount > 0: # #print(str(codes.totalCount) + " strategies found at " + currentSize) self.fnames, self.unique = single_loop(self, self.fnames, self.unique, self.cwd, codes) # else: # #print("0 strategies found for filesize " + currentSize) sleep(self.sleep_timer) break except Exception as e: # #print("total count threw an error. retrying") # #print(e) sleep(self.exception_sleep_timer) def single_loop(self, fnames, unique, cwd, codes): for i, code in enumerate(codes): for code_try in range(3): try: self.class_name_original = "" self.class_name_modified = "" self.strategyContent_original = "" self.strategyContent_modified_text = "" self.file_name_with_path_original = "" self.file_name_with_path_modified = "" self.class_name_original = "" self.class_name_modified = "" self.sha_original = "" self.sha_modified = "" self.strategyContent_original = code.decoded_content.decode() self.sha_original = md5.sha256(self.strategyContent_original.encode('utf-8')).hexdigest() if i == self.limit: break file_name_orig = code.download_url[code.download_url.rindex("/") + 1:] try: file_name_orig_without_extension = file_name_orig[:file_name_orig.rindex(".")] except: # did not find an extension, going with the file name directly... file_name_orig_without_extension = file_name_orig file_name = file_name_orig try: file_ext = file_name[file_name.rindex(".") + 1:] file_name = file_name[:file_name.rindex(".")] except: file_ext = "py" if file_name in fnames: duplicate = True if self.skiponduplicate: continue idx = 1 while True: file_name_with_idx = file_name + "_" + str(idx) if not file_name_with_idx + "." + file_ext in fnames: file_name = file_name_with_idx fnames.append(file_name + "." + file_ext) break else: idx += 1 else: duplicate = False fnames.append(file_name) pushed_at = code.repository.pushed_at self.class_name_modified = "github" + "_" + \ code.repository.full_name + "__" + \ file_name_orig_without_extension + "__" + \ str(pushed_at.year) + \ str(pushed_at.month).zfill(2) + \ str(pushed_at.day).zfill(2) + "_" + \ str(pushed_at.hour).zfill(2) + \ str(pushed_at.minute).zfill(2) + \ str(pushed_at.second).zfill(2) # modify file names to remove forbidden characters (like spaces or "(1)" ) self.class_name_modified = re.sub(r"[%]\d+", "", self.class_name_modified) self.class_name_modified = re.sub(r"[(]\d+[)]", "", self.class_name_modified) self.class_name_modified = self.class_name_modified.replace("(", "_") self.class_name_modified = self.class_name_modified.replace(")", "_") self.class_name_modified = self.class_name_modified.replace("-", "_") self.class_name_modified = self.class_name_modified.replace(".", "_") self.class_name_modified = self.class_name_modified.replace("/", "_") self.file_name_with_path_original = re.sub(r"[%]\d+", "", self.file_name_with_path_original) self.file_name_with_path_original = re.sub(r"[(]\d+[)]", "", self.file_name_with_path_original) # define file names self.file_name_with_path_original = self.dir_name_original + file_name + "." + file_ext self.file_name_with_path_modified = self.dir_name_modified + self.class_name_modified + ".py" # modify strategy content to include the source self.strategyContent_modified_text, self.class_name_original, self.class_name_modified = \ edit_strategy_content(code, self.strategyContent_original, self.class_name_modified) if self.sha_original not in unique: unique.append(self.sha_original) with open(self.dir_namex + "/processed.txt", "a") as output: output.write( self.file_name_with_path_original + "|" + code.download_url + "|" + self.sha_original + "\n") if "(IStrategy)" in self.strategyContent_original: if "def populate_buy_trend" in self.strategyContent_original or \ "def populate_entry_trend" in self.strategyContent_original: #print(str(datetime.now()) + ": added " + self.class_name_modified + ".py") with open(self.file_name_with_path_original, "w+") as t: t.write(self.strategyContent_original) with open(self.file_name_with_path_modified, "w+") as t: t.write(self.strategyContent_modified_text) else: # #print("Not a Freqtrade Strategy") continue # else: # #print("duplicate strategy " + self.file_name_with_path_original + " found, skipping save.") sleep(self.sleep_timer) break except Exception as e: # #print("error produced on single_loop, retry...") # #print(e) sleep(self.exception_sleep_timer) continue return fnames, unique if __name__ == "__main__": #print("Started the infinite loop!") while 1 == 1: # run forever! github_downloader.main(github_downloader)