Introduction à Python
  • Back to Main Website
  • Home
  • Comprendre Et Installer Python
    • Comprendre Et Installer Python
    • Histoire du Langage
    • Exécution d’un Programme Python
    • Versions et Compilation de Python
    • Le PATH
    • Path.. et environnements virtuels!
    • Les IDEs
    • Les Notebooks

    • Quelques IDEs en Python
    • VsCode - L’IDE Flexible et Polyvalent
    • Spyder - Un IDE Orienté Science des Données
    • PyCharm - L’IDE orienté Python

    • Travaux Pratiques
    • TP Guidé - Installer plusieurs versions de python avec pyenv
    • TP - Construire son python depuis la source
  • Syntaxes et Concepts de Base
    • Syntaxes et Concepts de Base
    • Syntaxe et objets de Base Python
    • Fonctions et Modules
    • Introduction à la POO en Python

    • Travaux Pratiques
    • Exercices d’applications
    • Base de la POO: Exercice
  • Les Librairies Python
    • Les Librairies Python
    • Installer et importer des librairies en Python
    • Les DataFrames
    • Exemple sur pandas
    • Calcul Scientifique et Optimization
    • Machine Learning
    • Recupérer des données du net
    • Python - Aussi un language pour servir des données
    • Visualiser et présenter ces données avec Python

    • Travaux Pratiques
    • TP-3 Libraries
    • TP - Utiliser pandas
  • Bonne pratiques, Dangers, et Astuces
    • Bonne pratiques, Dangers, et Astuces
    • Mutabilité et Scope
    • Typage en Python
    • Asynchronie et Multiprocessing

    • Travaux Pratiques
    • Modern Portfolio Theory - Practical Work
    • Modern Portfolio Theory - Practical Work - Corrected version
    • TP Python for Finance: Introduction to Option Pricing
    • TP Python for Finance: Introduction to Option Pricing - Corrected Version
    • TP - Creer un outil de récupération de donnée
  • Concepts avancés
    • Concepts avancés
    • L’arbre Syntaxique Abstrait ou AST
    • Python Orienté Objet - Les Dunders
    • Python Orienté Objet - les Design Patterns

    • Travaux Pratiques
    • TP-5
  • Sujets de Projets possibles
    • Projets
    • Projets Introduction à Python - Millésime 2024
    • Projets Introduction à Python - Millésime 2025
  • Code source
  1. Travaux Pratiques
  2. TP-5
  • Concepts avancés
  • L’arbre Syntaxique Abstrait ou AST
  • Python Orienté Objet - Les Dunders
  • Python Orienté Objet - les Design Patterns
  • Travaux Pratiques
    • TP-5

On this page

  • TP 5 : POO et Gestion de Portefeuille
    • Objectif Principal du TP
      • Tâches à Réaliser
  1. Travaux Pratiques
  2. TP-5

TP-5

Cours
Fondamentaux
Author

Remi Genet

Published

2025-02-12

Pyodide Status

Initializing Python Packages

TP 5 : POO et Gestion de Portefeuille


En reprenant le code du TP précédent pour obtenir des données:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
from typing import List, Any, Union
import time
import json
import pandas as pd
import time
from tqdm import tqdm
import asyncio
import aiohttp
import requests
from IPython.display import display
import multiprocessing
from functools import wraps, reduce
def in_notebook():
    try:
        from IPython import get_ipython
        if 'IPKernelApp' not in get_ipython().config:  # pragma: no cover
            return False
    except ImportError:
        return False
    except AttributeError:
        return False
    return True
def run_in_process(async_func):
    # if in_notebook():
    #     return async_func
    @wraps(async_func)
    def wrapper(*args, **kwargs):
        def run():
            loop = asyncio.new_event_loop()
            asyncio.run(async_func(*args, **kwargs))
        process = multiprocessing.Process(target=run)  
        process.start() 
        process.join()    
    return wrapper
def sync_wrap(func):
    @wraps(func)
    async def wrapped_func(*args, **kwargs):
        loop = asyncio.get_event_loop()
        f = lambda : func(*args,  **kwargs)
        return await loop.run_in_executor(None, f)
    return wrapped_func
class DataGatherer:
    interval='1m'
    def __init__(self, symbol: str, start_time: str = '20230101', end_time: str = '20230131', cache_folder=''):
        self.symbol = symbol
        self.start_time = pd.Timestamp(start_time, tz = 'UTC')
        self.end_time = pd.Timestamp(end_time, tz = 'UTC')
        delta = (self.end_time - self.start_time)
        n_period = int(delta.total_seconds() / 60 / 1000) + 1
        self.start_times = [self.start_time + pd.Timedelta(minutes=1000) * iter for iter in range(n_period)]
        
        
    def _load_underlying_data(self):
        # load all the bars
        pass
    
    def _make_url(self, start_time: pd.Timestamp):
        pass 
    
    def _get_data(self):
        pass
    
    
    @staticmethod
    def group_df(df: pd.DataFrame, col: str, group_size: str = '1H') -> pd.DataFrame:
        """
        Groups the DataFrame based on the specified time interval and aggregates the data.
        The function supports grouping by hours ('H'), minutes ('m'), or days ('d'). It aggregates
        the data to get the first value of 'open', maximum of 'high', minimum of 'low', last value of 
        'close', and the sum of 'volume' for each group.
        Args:
        df (pd.DataFrame): The DataFrame to be grouped.
        col (str): The column name in the DataFrame that contains the datetime information.
        group_size (str): The size of the grouping interval, e.g., '1H', '30m', '1d'.
        Returns:
        pd.DataFrame: A DataFrame grouped and aggregated based on the specified interval.
        Raises:
        ValueError: If the group size is not defined properly.
        """
        if group_size[-1] == 'H':
            df['group'] = df[col].apply(lambda x: x.replace(hour=x.hour - x.hour % int(group_size[:-1]), minute=0, second=0))
        elif group_size[-1] == 'm':
            df['group'] = df[col].apply(lambda x: x.replace(minute=x.minute - x.minute % int(group_size[:-1]), second=0))
        elif group_size[-1] == 'd':
            df['group'] = df[col].apply(lambda x: x.replace(day=x.day - x.day % int(group_size[:-1]), hour=0, minute=0, second=0))
        else:
            raise ValueError("Undefined group size")
        return df.groupby('group').agg({'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum', 'quote asset volume': 'sum'})
    
    def get_data(self, start_time: str = None, end_time: str = None, interval: str = None):
        start_time = start_time or self.start_time
        end_time = end_time or self.end_time
        interval = interval or self.interval
        view = self.dataframe[self.dataframe['open time'].apply(lambda x: x<end_time and x>start_time)]
        return DataGatherer.group_df(view, 'open time', interval)
        
    
class BinanceSpotDataGatherer(DataGatherer):
    url = "https://api.binance.com/api/v3/klines"
    
        
    def make_url(self, start_time):
        limit = min(1000, (self.end_time - start_time).value)
        return f'https://api.binance.com/api/v3/klines?symbol={self.symbol}&interval={self.interval}&startTime={start_time}&limit=1000'
    
    def _load_underlying_data(self):
        all_data = []
        for start in tqdm(self.start_times):
            data = self._get_data(self._make_url(start))
            all_data += data
        self.dataframe = BinanceSpotDataGatherer.data_to_df(all_data)
        
    async def _aload_underlying_data(self):
        coros = []
        for start in tqdm(self.start_times):
            coros.append(self._aget_data(self._make_url(start)))
        datas = await asyncio.gather(*coros)
        all_data = []
        for data in datas:
            all_data += data
        self.dataframe = BinanceSpotDataGatherer.data_to_df(all_data)
        
    
    def _make_url(self, start_time: pd.Timestamp):
        limit = min(1000, int((self.end_time - self.start_time).total_seconds() / 60) + 1)
        return f'https://api.binance.com/api/v3/klines?symbol={self.symbol}&startTime={int(start_time.timestamp() * 1000)}&interval={self.interval}&limit={limit}' 
    
    def _get_data(self, url, max_retry = 4, current_retry = 0):
        if current_retry >= max_retry:
            raise ValueError('Too much error')
        response = requests.get(url)
        if response.status_code == 200:
            return response.json()
        elif response.status_code == 429:
            time.sleep(10)
            return self._get_data(url, max_retry=max_retry, current_retry=current_retry+1)
        else:
            raise ValueError(f'Unknow status code {response.status_code}')
        
    async def _aget_data(self, url):
        async with aiohttp.ClientSession() as session:
            try:
                async with session.get(url) as response:
                    response.raise_for_status()
                    data_asynchrone = await response.json()
                    return data_asynchrone 
            except Exception as e:
                raise e
    
    
    @staticmethod
    def data_to_df(data):
        df = pd.DataFrame(
        data,
            columns=[
                "Open time",
                "Open",
                "High",
                "Low",
                "Close",
                "Volume",
                "Close time",
                "Quote asset volume",
                "Number of trades",
                "Taker buy base asset volume",
                "Taker buy quote asset volume",
                "Ignore",
            ],
        ).astype(float)
        df["Open time"] = (
            pd.to_datetime(df["Open time"], unit="ms")
            .dt.tz_localize("UTC")
            .dt.floor("us")
        )
        df["Close time"] = (
            pd.to_datetime(df["Close time"], unit="ms")
            .dt.tz_localize("UTC")
            .dt.floor("us")
        )
        df.columns = [c.lower() for c in df.columns]
        df = df.drop_duplicates()
        return df
    
class BinanceSpotDataGathererSmart(BinanceSpotDataGatherer):
    
    async def _load_underlying_data(self):
        self.dataframe = BinanceSpotDataGatherer.data_to_df(
            reduce(lambda x, y: list(x) + list(y),
                   await asyncio.gather(*[
                       self._get_data(
                           self._make_url(start)
                       ) for start in self.start_times
                   ])))
    
    @sync_wrap
    def _get_data(self, url, max_retry = 4, current_retry = 0):
        if current_retry >= max_retry:
            raise ValueError('Too much error')
        response = requests.get(url)
        used_weight = int(response.headers['x-mbx-used-weight-1m'])
        if used_weight > 800:
            print(f'warning: {used_weight} used_weight!')
        if response.status_code == 200:
            return response.json()
        elif response.status_code == 429:
            print('warning, received a 429 error - used_weight= {used_weight} - retrying in 60 seconds')
            time.sleep(60)
            return self._get_data(url, max_retry=max_retry, current_retry=current_retry+1)
        else:
            raise ValueError(f'Unknow status code {response.status_code}')
        
@run_in_process
async def test():
    t1 = time.time()
    btc_loader = BinanceSpotDataGathererSmart('BTCUSDT')
    await btc_loader._load_underlying_data()
    print(time.time()-t1)
    display(btc_loader.get_data(interval='1H'))
    t1 = time.time()
    eth_loader = BinanceSpotDataGathererSmart('ETHUSDT')
    await eth_loader._aload_underlying_data()
    print(time.time()-t1)
    display(eth_loader.get_data(interval='1H'))
test()

Objectif Principal du TP

L’objectif de ce TP est d’appliquer les compétences en programmation Python et en analyse financière pour la récupération, l’analyse et la gestion de données de marché. Utilisant la classe développée dans la séance précédente, les étudiants effectueront des calculs statistiques et construiront un simulateur de portefeuille d’investissement.

Tâches à Réaliser

  1. Utilisation de Pandas pour Calculer la Volatilité :
    • Appliquer la méthode rolling de Pandas pour calculer la volatilité des actifs individuels. Les étudiants utiliseront les données récupérées via la classe BinanceSpotDataGathererSmart pour calculer la volatilité sur une fenêtre temporelle définie.
  2. Combinaison des DataFrames de Volatilité :
    • Fusionner les différentes DataFrames contenant les volatilités calculées pour chaque actif. Cette étape est cruciale pour la comparaison et l’analyse croisée des différentes volatilités d’actifs.
  3. Calcul de la Matrice de Covariance et du Rendement Moyen :
    • Utiliser les données combinées pour calculer la matrice de covariance entre les actifs. Cette matrice est essentielle pour comprendre la relation et la corrélation entre différents investissements.
    • Calculer également le rendement moyen des différents actifs pour avoir une vue d’ensemble de la performance.
  4. Création d’une Classe Portefeuille :
    • Développer une classe Portefeuille qui permet de gérer des investissements en dollars ($). Cette classe doit pouvoir contenir les proportions de plusieurs actifs, et implementer des méthodes afin de calculer la volatilité, le rendement, et le ratio de Sharpe du portefeuille à partir de la matrice de covariance.
  5. Implémentation des Méthodes de Gestion de Portefeuille :
    • Implémenter les méthodes magiques (__add__, etc.) pour permettre la combinaison de différents portefeuilles.
    • Implementer __str__ our __repr__ pour afficher les informations du portefeuille.
Back to top
Python Orienté Objet - les Design Patterns

Introduction à Python, Rémi Genet.
Licence
Code source disponible sur Github

 

Site construit avec et Quarto
Inspiration pour la mise en forme du site ici
Code source disponible sur GitHub