Bourse/BourseV4/Model/Donnee.py

from Config.paths import SysExploite

if SysExploite == "linux" :
	from Model.My_Redis import *
	
from Model.Action import *
from Config.DEBUG import *
from Model.profil import profile

import yfinance as yf
import pandas as pd
import numpy as np
from svgelements import Close
from ta import add_all_ta_features
from datetime import datetime, timedelta, date
import json
from bisect import bisect_left
import pickle


def to_date(x):
	if isinstance(x, str):
		return datetime.strptime(x, "%Y-%m-%d").date()
	return x


class CL_Donnee:
	def __init__(self, date_debut, nb_jours, Parametre):
		self.valeurCloture = {}
		self.listeDonne = {}
		self.listeAction = listeAction()
		if DEBUGpickle == 1 :
			UUID = Parametre.UUIDRAM(date_debut,nb_jours)
			pickle_path = f"RAM_cache_{UUID}.pkl"
			
			if not os.path.exists(pickle_path):
				with open(pickle_path, "wb") as f:
					self.RAMDonne(date_debut, nb_jours, Parametre)
					data = {"valeurCloture": self.valeurCloture,"listeDonne": self.listeDonne}
					pickle.dump(data, f)
			else :
				with open(pickle_path, "rb") as f:
					data = pickle.load(f)
				
				self.valeurCloture = data["valeurCloture"]
				self.listeDonne = data["listeDonne"]
			
		else:
			self.RAMDonne(date_debut, nb_jours, Parametre)
		self.Preprocess()
	
	@profile
	def RAMDonne(self, date_debut, nb_jours, Parametre):
		
		date_depart = datetime.strptime(date_debut, "%Y-%m-%d").date() - timedelta(days=Parametre.NbJourRecul * 2)
		for action in self.listeAction:
			self.listeDonne[action] = {}
			self.valeurCloture[action] = {}
			listejour = ListeJour(date_depart, nb_jours + Parametre.NbJourRecul * 2)
			for date in listejour:
				valeur = LireRedis(action, date)
				if valeur is not None:
					self.listeDonne[action][date] = json.loads(valeur)
					self.valeurCloture[action][date] = json.loads(valeur).get('Close')
	
	@profile
	def ValeurDate(self, Action, Date):
		return self.valeurCloture[Action][Date]
	
	@profile
	def ResultatGlobal(self, Date, nbJour):
		listeJour = ListeJour(Date, nbJour)
		Open = 0
		Close = 0
		for action in self.listeAction:
			Open += self.LireUneDonnee(action, listeJour[0]).get('Open')
			Close += self.LireUneDonnee(action, listeJour[-1]).get('Close')
		return (Close) / Open
	
	@profile
	def Preprocess(self):
		self.dataTriee = {}
		
		for action, data in self.listeDonne.items():
			jours = [to_date(d) for d in data.keys()]
			valeurs = list(data.values())
			
			tri = sorted(zip(jours, valeurs))
			
			self.dataTriee[action] = {
				"jours": [x[0] for x in tri],
				"valeurs": [x[1] for x in tri]
			}
	
	@profile
	def LireUneDonnee(self, Action, Date):
		cible = to_date(Date)
		
		data = self.dataTriee[Action]
		jours = data["jours"]
		valeurs = data["valeurs"]
		
		i = bisect_left(jours, cible)
		
		if i >= len(jours):
			return valeurs[-1]
		
		return valeurs[i]
	
	@profile
	def LireUneDonneeTry(self, Action, Date):
		try:
			return self.listeDonne[Action][Date]
		except (KeyError, IndexError, TypeError):
			return None
	
	@profile
	def DecalerJour(self, Date, Horizon):
		date_obj = datetime.strptime(Date, "%Y-%m-%d")
		nouvelle_date = date_obj + timedelta(days=Horizon)
		return nouvelle_date.strftime("%Y-%m-%d")
	
	@profile
	def PreparationDonnee(self, Date, nbJourRecule, Analysis, Horizon, ListeActions):
		
		listeJour = self.ListeJourInverse(Date, nbJourRecule)
		
		ListeReturn = []
		rendement_futur = []
		for action in ListeActions:
			
			# ===== HISTORIQUE (comme avant) =====
			for jour in listeJour:
				res = self.LireUneDonnee(action, jour)
				for param in Analysis:
					ListeReturn.append(res.get(param))
			
			# ===== TARGET (nouveau) =====
			prix_actuel = self.LireUneDonnee(action, Date).get("Close")
			prix_futur = self.LireUneDonnee(action, self.DecalerJour(Date, Horizon)).get("Close")
			
			rendement = (prix_futur - prix_actuel) / prix_actuel
			rendement_futur.append(rendement)
		
		# ===== reshape historique =====
		historique_indicateurs = np.array(ListeReturn).reshape(
			(1, len(ListeActions), nbJourRecule, len(Analysis))
		)
		
		# ===== normalisation =====
		mean = historique_indicateurs.mean(axis=(0, 2), keepdims=True)
		std = historique_indicateurs.std(axis=(0, 2), keepdims=True)
		historique_indicateurs = (historique_indicateurs - mean) / (std + 1e-8)
		
		# ===== reshape target =====
		rendement_futur = np.array(rendement_futur).reshape((1, len(ListeActions)))
		
		return historique_indicateurs, rendement_futur
	
	def PreparationJours(self, StartDay, NBJoursEntrainement, Parametre, Var_listeAction):
		liste_Entree = []
		liste_Target = []
		listeJours = ListeJour(StartDay, NBJoursEntrainement)
		for jour in listeJours:
			Entree, target = self.PreparationDonnee(jour, Parametre.NbJourRecul, Parametre.Analysis, Parametre.Horizon,
			                                        Var_listeAction)
			Entree = np.asarray(Entree)
			target = np.asarray(target)
			if Entree.ndim == 4 and Entree.shape[0] == 1:
				Entree = Entree[0]
			liste_Entree.append(Entree)
			liste_Target.append(target)
		Entree = np.stack(liste_Entree, axis=0)
		target = np.stack(liste_Target, axis=0)
		target = target.reshape(target.shape[0], target.shape[2])
		resGlobal = self.ResultatGlobal(StartDay, NBJoursEntrainement)
		return Entree, target, resGlobal, listeJours
	
	@profile
	def ListeJourInverse(self, Date, NbJour):
		liste = []
		date_obj = datetime.strptime(Date, "%Y-%m-%d").date()
		i = 1
		while (i < NbJour + 1):
			date_obj = date_obj + timedelta(days=-1)
			valeur1 = self.LireUneDonneeTry("MT.AS", date_obj.strftime("%Y-%m-%d"))
			valeur2 = self.LireUneDonneeTry("AC.PA", date_obj.strftime("%Y-%m-%d"))
			if valeur1 is not None and valeur2 is not None:
				liste.insert(0, date_obj.strftime("%Y-%m-%d"))
				i = i + 1
		return liste
	
	def DataPickelData(self, StartDay, NBJoursEntrainement, Parametre, Var_listeAction):
		pickle_path = f"data_cache_{Parametre.UUIDData(StartDay, NBJoursEntrainement)}.pkl"
		if os.path.exists(pickle_path):
			with open(pickle_path, "rb") as f:
				Entree, Target, resGlobal, listeJours = pickle.load(f)
		else:
			Entree, Target, resGlobal, listeJours = self.PreparationJours(StartDay, NBJoursEntrainement, Parametre,
			                                                                Var_listeAction)
			with open(pickle_path, "wb") as f:
				pickle.dump((Entree, Target, resGlobal, listeJours), f)
		return Entree, Target, resGlobal, listeJours