Closed JoeGithub53 closed 2 months ago
Nope ist bisher nicht möglich
zunächst einmal möchte ich mich für die großartige Arbeit bedanken, die Du bisher mit der Wikifolio API geleistet hast. Die Implementierung in TopRoboFolio hat es mir ermöglicht, innovative und teilautomatisierte Handelsstrategien zu entwickeln, die jeden der es möchte, einen großen Mehrwert bieten. Die ersten Interessierten, waren damit bereits so erfolgreich, daß sie bereits in den ersten 21 Tagen den Investierstatus beantragen konnten. Ich selber wurde dadurch auch überrascht, weil ab da die teilautomatisierte Unterstützung durch TopRoboFolio nicht mehr funktioniert. Die Anwender muessen auf Clientkey und Userkey umstellen. Da wäre die Möglichkeit, Limit Kauf/Verkäufe über die API durchzuführen, von entscheidender Bedeutung. Ich wäre Dir sehr dankbar, wenn Dudie Implementierung dieser Funktion in Erwägung ziehen könntest. Wenn es irgendwelche Ressourcen, Tests oder Feedback von meiner Seite benötigt, um diesen Prozess zu unterstützen, stehen ich Dir gerne zur Verfügung.
habe versucht folgende def_buy unterroutine einzubauen: def buy_limit(self, wikifolio_id: str, amount: int, isin: str, valid_until: str, limit_price: float): """Place a buy limit order.
Args:
wikifolio_id (str): The ID of the wikifolio.
amount (int): The amount of the underlying asset to buy.
isin (str): The ISIN of the underlying asset.
valid_until (str): The valid until date in the format 'YYYY-MM-DD'.
limit_price (float): The limit price for the order.
Returns:
dict: The response from the API.
"""
if not self.sessionToken:
raise ValueError("Session token must be set before making a request")
headers = self.headers
payload = {
"wikifolioSymbol": wikifolio_id,
"underlying": isin,
"amount": amount,
"orderType": "BuyLimit",
"validUntilDate": valid_until,
"limitPrice": limit_price
}
logger.info(f'Placing buy limit order for wikifolio {wikifolio_id}')
r = requests.post(
"https://trading-api.wikifolio.com/v1/orders",
headers=headers,
json=payload
)
r.raise_for_status()
logger.debug(f'Order response: {r.json()}')
return r.json() Beim Test mit folgendem Aufruf kommt folgender Fehler: from wikifolioTradingAPI import WikifolioTradingAPI
wf_api = WikifolioTradingAPI("mein_client_key", "mein_user_key") print(wf_api.list_wikifolios()) print(wf_api.get_wikifolio('wf0tmastba'))
wf_api.buy_limit(wikifolio_id="wf0tmastba", isin="DE0005493365", amount=1, valid_until="2024-07-26", limit_price=150.0) [2024-07-26 20:05:36] INFO Placing buy limit order for wikifolio wf0tmastba
Traceback (most recent call last):
File "d:#flask_TopRoboFolio_ProjektV3\wikifolioTradingAPI\Test_neu.py", line 8, in
Habe eine Limit Order mit folgendem Code geschafft:
import dataclasses
import requests
from dacite import from_dict
from typing import List, Optional, Tuple
import logging
import coloredlogs
from classes.WikifolioListItem import WikifolioListItem from classes.Wikifolio import Wikifolio from classes.Underlying import Underlying from classes.OrderStatusResponse import OrderStatusResponse import json
logger = logging.getLogger(name) coloredlogs.install(level='INFO', logger=logger, fmt='[%(asctime)s] %(levelname)s %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
@dataclasses.dataclass class WikifolioTradingAPI: clientApiKey: str userApiKey: str
def __post_init__(self) -> None:
headers = {
'accept': 'application/json',
'clientApiKey': self.clientApiKey,
'userApiKey': self.userApiKey
}
logger.info('Getting session token')
r = requests.post('https://trading-api.wikifolio.com/v1/sessions', headers=headers)
r.raise_for_status()
logger.debug(f'Session token: {r.json()["sessionToken"]}')
self.sessionToken = r.json()['sessionToken']
self.headers = {'accept': 'application/json', 'sessionToken': self.sessionToken}
def list_wikifolios(self) -> List[WikifolioListItem]:
"""List wikifolios
Returns:
List[WikifolioListItem]: A list of wikifolios
"""
headers = {
'accept': 'application/json',
'sessionToken': self.sessionToken
}
params = {
'pageNumber': 1
}
list_wikifolios = []
logger.info(f'Getting wikifolios, page {params["pageNumber"]}')
r = requests.get('https://trading-api.wikifolio.com/v1/wikifolios', headers=headers, params=params)
r.raise_for_status()
logger.info(f'Found {r.json()["totalPages"]} pages of wikifolios')
logger.info(f'Found {len(r.json()["results"])} wikifolios on page {params["pageNumber"]}')
list_wikifolios += r.json()['results']
while r.json()['pageNumber'] < r.json()['totalPages']:
params['pageNumber'] += 1
r = requests.get('https://trading-api.wikifolio.com/v1/wikifolios', headers=headers, params=params)
r.raise_for_status()
logger.info(f'Found {len(r.json()["results"])} wikifolios on page {params["pageNumber"]}')
list_wikifolios += r.json()['results']
return [from_dict(data_class=WikifolioListItem, data=wikifolio) for wikifolio in list_wikifolios]
def get_wikifolio(self, wikifolioSymbol: str) -> Wikifolio:
"""Get a wikifolio
Args:
wikifolioSymbol (str): The wikifolio symbol
Returns:
Wikifolio: The wikifolio
"""
headers = {
'accept': 'application/json',
'sessionToken': self.sessionToken
}
logger.info(f'Getting wikifolio {wikifolioSymbol}')
r = requests.get(f'https://trading-api.wikifolio.com/v1/wikifolios/{wikifolioSymbol}', headers=headers)
r.raise_for_status()
return from_dict(data_class=Wikifolio, data=r.json())
def list_wikifolio_underlyings(self, wikifolioSymbol: str) -> List[Underlying]:
"""List underlyings for a wikifolio
Args:
wikifolioSymbol (str): The wikifolio symbol
Returns:
List[Underlying]: A list of underlyings
"""
headers = {
'accept': 'application/json',
'sessionToken': self.sessionToken
}
params = {
'pageNumber': 1
}
list_underlyings = []
logger.info(f'Getting underlyings for {wikifolioSymbol}, page {params["pageNumber"]}')
r = requests.get(f'https://trading-api.wikifolio.com/v1/wikifolios/{wikifolioSymbol}/underlyings', headers=headers, params=params)
r.raise_for_status()
logger.info(f'Found {r.json()["totalPages"]} pages of underlyings for {wikifolioSymbol}')
logger.info(f'Found {len(r.json()["results"])} underlyings for {wikifolioSymbol} on page {params["pageNumber"]}')
list_underlyings += r.json()['results']
while r.json()['pageNumber'] < r.json()['totalPages']:
params['pageNumber'] += 1
logger.info(f'Getting underlyings for {wikifolioSymbol}, page {params["pageNumber"]}')
r = requests.get(f'https://trading-api.wikifolio.com/v1/wikifolios/{wikifolioSymbol}/underlyings', headers=headers, params=params)
r.raise_for_status()
logger.info(f'Found {len(r.json()["results"])} underlyings for {wikifolioSymbol} on page {params["pageNumber"]}')
list_underlyings += r.json()['results']
return [from_dict(data_class=Underlying, data=underlying) for underlying in list_underlyings]
def list_wikifolio_orders(self, wikifolioSymbol: str, status: Optional[str] = None) -> List[OrderStatusResponse]:
"""List orders for a wikifolio
Args:
wikifolioSymbol (str): The wikifolio symbol
status (Optional[str]): The order status to filter by. Defaults to None. Possible values: 'Inactive', 'Waiting', 'Active', 'Evaluating', 'Executing', 'RequestingExecutionInformation', 'PartiallyExecutedActive', 'Executed', 'PartiallyExecutedExecuted', 'Deleted', 'DeleteRequested', 'Updated', 'Obsolete', 'Error', 'Rejected', 'Undone', 'Abandoned'
Returns:
List[OrderStatusResponse]: A list of orders
"""
headers = {
'accept': 'application/json',
'sessionToken': self.sessionToken
}
params = {
'pageNumber': 1
}
if status:
logger.debug(f'Filtering orders by status: {status}')
if status in ['Inactive', 'Waiting', 'Active', 'Evaluating', 'Executing', 'RequestingExecutionInformation', 'PartiallyExecutedActive', 'Executed', 'PartiallyExecutedExecuted', 'Deleted', 'DeleteRequested', 'Updated', 'Obsolete', 'Error', 'Rejected', 'Undone', 'Abandoned']:
params['status'] = status
else:
logger.error(f'Invalid order status: {status}, ignoring filter')
del params['status']
list_orders = []
logger.info(f'Getting orders for {wikifolioSymbol}, page {params["pageNumber"]}')
r = requests.get(f'https://trading-api.wikifolio.com/v1/wikifolios/{wikifolioSymbol}/orders', headers=headers, params=params)
r.raise_for_status()
logger.info(f'Found {r.json()["totalPages"]} pages of orders for {wikifolioSymbol}')
logger.info(f'Found {len(r.json()["results"])} orders for {wikifolioSymbol} on page {params["pageNumber"]}')
list_orders += r.json()['results']
while r.json()['pageNumber'] < r.json()['totalPages']:
params['pageNumber'] += 1
logger.info(f'Getting orders for {wikifolioSymbol}, page {params["pageNumber"]}')
r = requests.get(f'https://trading-api.wikifolio.com/v1/wikifolios/{wikifolioSymbol}/orders', headers=headers, params=params)
r.raise_for_status()
logger.info(f'Found {len(r.json()["results"])} orders for {wikifolioSymbol} on page {params["pageNumber"]}')
list_orders += r.json()['results']
return [from_dict(data_class=OrderStatusResponse, data=order) for order in list_orders]
def buy_limit(self, wikifolio_id: str, amount: int, isin: str, valid_until: str, limit_price: float):
"""Place a buy limit order.
Args:
wikifolio_id (str): The ID of the wikifolio.
amount (int): The amount of the underlying asset to buy.
isin (str): The ISIN of the underlying asset.
valid_until (str): The valid until date in the format 'YYYY-MM-DD'.
limit_price (float): The limit price for the order.
Returns:
dict: The response from the API.
"""
if not self.sessionToken:
raise ValueError("Session token must be set before making a request")
#headers = self.headers
# Definiere die Headers mit sessionToken
headers = {
'Content-Type': 'application/json',
'accept': 'application/json',
'sessionToken': self.sessionToken # Ersetze self.sessionToken durch deinen tatsächlichen Token
}
# Erstelle ein Dictionary mit den Variablen
params = {
'wikifolioSymbol': wikifolio_id,
'underlying': isin,
'amount': amount,
'orderType': 'BuyLimit',
'validUntilDate': valid_until,
'limitPrice': limit_price
#'stopPrice': stop_price
}
# Konvertiere das Dictionary in einen JSON-String
json_params = json.dumps(params, indent=2)
print(json_params)
# Definiere die URL
url = 'https://trading-api.wikifolio.com/v1/limitorders'
#print("Payload:", payload)
logger.info(f'Placing buy limit order for wikifolio {wikifolio_id}')
# Sende die Anfrage
response = requests.post(url, headers=headers, data=json_params)
# Überprüfe die Antwort
try:
response.raise_for_status()
print("Order placed successfully")
except requests.exceptions.HTTPError as err:
print(f"HTTP error occurred: {err}")
print(f"Response content: {response.content}")
Aufruf über meint Test.py: from wikifolioTradingAPI import WikifolioTradingAPI
wf_api = WikifolioTradingAPI("mein_Client_Key", "mein_User_Key") wf_api.buy_limit("wf0tmastba", 1, "DE0005493365", "2024-07-26", 150.0)
Ergebnis: { "wikifolioSymbol": "wf0tmastba", "underlying": "DE0005493365", "amount": 1, "orderType": "BuyLimit", "validUntilDate": "2024-07-26", "limitPrice": 150.0 } [2024-07-26 22:52:07] INFO Placing buy limit order for wikifolio wf0tmastba Order placed successfully
Danke ich gucke mir das bei Gelegenheit an
danke, da machst Du mir und den zukünftigen Nutzern eine große Freude. Es eilt auch nicht, ich kann ja schon mal meine Variante in meine Programme einbauen, auch wenn es noch nicht besonders gut programmiert ist. Dafür freue ich mich um so mehr auf Deine Variante!
Also, ich habe Limit und Quote Orders direkt wie in der API eingebaut. Funktioniert bei mir super, unterstützt auch StopLoss und den Verkauf von Wertpapieren
Ich bin begeistert in welcher kurzen Zeit und Qualität Du das wieder gelöst hast. Ich habe auch gleich mit GitGUI geklont und buy, sell und delete order getest. Alles funktioniert auf Anhieb perfekt. So kann ich diese tolle Lösung gleich diese Woche in meine Anwendung einbauen und nun endlich meine Wikifolios halb automatisiert pflegen. Für Deine tolle Arbeit möchte ich Dir nochmals ganz herzlich danken. Und wenn meine zukünftigen Kunden für die ich meine Anwendung ebenfalls wie Du kostenlos anbiete, auch so begeistert sind, werde ich Dich immer erwähnen.
ist der Limit Kauf/Verkauf bereits mit der neuen trading API möglich?. Ich bräuchte diese Funktion, weil meine erste TopRoboFolio Kundin in 21 Tagen ihr Wikifolio investierbar beauftragen konnte und nun zur Nutzung auf Client und Userkey umstellen muesste.