A real-time OHLC (Open, High, Low, Close) generator from 1-minute Last Traded Price (LTP) data in Python
To create a real-time OHLC (Open, High, Low, Close) generator from 1-minute Last Traded Price (LTP) data in Python, we need to continuously fetch the LTP data, update our OHLC values, and store or display them in real-time.
#### Importing Libraries
```python
import pandas as pd
import numpy as np
import time
from datetime import datetime, timedelta
```
- **pandas**: Used for data manipulation and analysis.
- **numpy**: Used for numerical operations.
- **time**: Used for handling time-related tasks.
- **datetime**: Used for manipulating dates and times.
#### Class Definition
```python
class RealTimeOHLC:
def __init__(self):
self.data = pd.DataFrame(columns=['timestamp', 'ltp'])
self.data['timestamp'] = pd.to_datetime(self.data['timestamp'])
self.data['ltp'] = pd.to_numeric(self.data['ltp'])
self.current_ohlc = {'open': None, 'high': None, 'low': None, 'close': None}
self.current_minute = None
```
- **RealTimeOHLC class**: Defines the structure and behavior for generating real-time OHLC data.
- **__init__ method**: Initializes the class instance.
- **self.data**: A pandas DataFrame with columns `timestamp` and `ltp` (Last Traded Price). The `timestamp` column is converted to datetime format and `ltp` to numeric to ensure correct data types.
- **self.current_ohlc**: A dictionary to store the current minute's OHLC (Open, High, Low, Close) values.
- **self.current_minute**: Tracks the current minute to determine when to update OHLC values.
#### Fetching LTP
```python
def fetch_ltp(self):
"""Simulate fetching the last traded price from an API."""
ltp = np.random.uniform(100, 200) # Random price between 100 and 200
return ltp
```
- **fetch_ltp method**: Simulates fetching the Last Traded Price (LTP) from an API.
- Generates a random float between 100 and 200 to simulate the LTP.
#### Updating OHLC
```python
def update_ohlc(self, timestamp, ltp):
"""Update the OHLC values with the new LTP."""
if self.current_minute != timestamp.minute:
if self.current_minute is not None:
print(f"{self.current_minute}: {self.current_ohlc}")
self.current_minute = timestamp.minute
self.current_ohlc = {
'open': ltp,
'high': ltp,
'low': ltp,
'close': ltp
}
else:
self.current_ohlc['high'] = max(self.current_ohlc['high'], ltp)
self.current_ohlc['low'] = min(self.current_ohlc['low'], ltp)
self.current_ohlc['close'] = ltp
```
- **update_ohlc method**: Updates the OHLC values based on the new LTP.
- **if self.current_minute != timestamp.minute**: Checks if the current minute has changed.
- If it has, prints the OHLC values for the previous minute.
- Updates `self.current_minute` to the new minute and sets the initial OHLC values (`open`, `high`, `low`, and `close`) to the current LTP.
- **else**: If the minute has not changed, updates the `high`, `low`, and `close` values accordingly.
#### Running the Updater
```python
def run(self):
"""Run the real-time OHLC updater."""
while True:
timestamp = datetime.now()
ltp = self.fetch_ltp()
new_row = pd.DataFrame({'timestamp': [timestamp], 'ltp': [ltp]})
self.data = pd.concat([self.data, new_row], ignore_index=True)
self.update_ohlc(timestamp, ltp)
time.sleep(1) # Fetch new data every second (simulating real-time)
```
- **run method**: Runs the real-time OHLC updater in an infinite loop.
- **while True**: An infinite loop to keep fetching and processing data.
- **timestamp = datetime.now()**: Gets the current timestamp.
- **ltp = self.fetch_ltp()**: Fetches the simulated LTP.
- **new_row = pd.DataFrame({'timestamp': [timestamp], 'ltp': [ltp]})**: Creates a new DataFrame row with the current timestamp and LTP.
- **self.data = pd.concat([self.data, new_row], ignore_index=True)**: Appends the new row to the existing DataFrame.
- **self.update_ohlc(timestamp, ltp)**: Updates the OHLC values with the new LTP.
- **time.sleep(1)**: Waits for 1 second before fetching the next LTP, simulating real-time data fetching.
#### Main Execution
```python
if __name__ == "__main__":
ohlc_generator = RealTimeOHLC()
ohlc_generator.run()
```
- **Main Execution Block**: Creates an instance of `RealTimeOHLC` and runs the updater.
### Summary
This script simulates a real-time OHLC generator for stock trading data using random LTP values. The key functionalities include fetching simulated LTP values, updating OHLC values, and continuously processing data in real-time. In a real-world scenario, the `fetch_ltp` method would be replaced with an actual API call to fetch real-time trading data.
No Comments have been Posted.