59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
|
import requests
|
||
|
|
from datetime import datetime
|
||
|
|
from typing import List
|
||
|
|
from .base import BaseExchange, Trade
|
||
|
|
|
||
|
|
class LSExchange(BaseExchange):
|
||
|
|
@property
|
||
|
|
def name(self) -> str:
|
||
|
|
return "LS"
|
||
|
|
|
||
|
|
def fetch_latest_trades(self) -> List[Trade]:
|
||
|
|
# Today's trades endpoint
|
||
|
|
url = "https://www.ls-x.de/_rpc/json/.lstc/instrument/list/lstctradestoday"
|
||
|
|
|
||
|
|
# We might need headers to mimic a browser or handle disclaimer
|
||
|
|
headers = {
|
||
|
|
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
|
||
|
|
'Accept': 'application/json',
|
||
|
|
'Referer': 'https://www.ls-tc.de/'
|
||
|
|
}
|
||
|
|
|
||
|
|
try:
|
||
|
|
response = requests.get(url, headers=headers)
|
||
|
|
response.raise_for_status()
|
||
|
|
|
||
|
|
import csv
|
||
|
|
import io
|
||
|
|
f = io.StringIO(response.text)
|
||
|
|
# Header: isin;displayName;tradeTime;price;currency;size;orderId
|
||
|
|
reader = csv.DictReader(f, delimiter=';')
|
||
|
|
|
||
|
|
trades = []
|
||
|
|
for item in reader:
|
||
|
|
try:
|
||
|
|
price = float(item['price'].replace(',', '.'))
|
||
|
|
quantity = float(item['size'].replace(',', '.'))
|
||
|
|
isin = item['isin']
|
||
|
|
symbol = item['displayName']
|
||
|
|
time_str = item['tradeTime']
|
||
|
|
|
||
|
|
# Format: 2026-01-23T07:30:00.992000Z
|
||
|
|
ts_str = time_str.replace('Z', '+00:00')
|
||
|
|
timestamp = datetime.fromisoformat(ts_str)
|
||
|
|
|
||
|
|
trades.append(Trade(
|
||
|
|
exchange=self.name,
|
||
|
|
symbol=symbol,
|
||
|
|
isin=isin,
|
||
|
|
price=price,
|
||
|
|
quantity=quantity,
|
||
|
|
timestamp=timestamp
|
||
|
|
))
|
||
|
|
except Exception:
|
||
|
|
continue
|
||
|
|
return trades
|
||
|
|
except Exception as e:
|
||
|
|
print(f"Error fetching LS data: {e}")
|
||
|
|
return []
|