-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathAlphaInsider.py
114 lines (95 loc) · 3.25 KB
/
AlphaInsider.py
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
import requests
import json
import os
from dotenv import load_dotenv
# Load API Key from .env file
load_dotenv()
api_key = os.getenv('ALPHAINSIDER_API')
# Define the API endpoints and parameters
BASE_URL = "https://alphainsider.com/api/"
HEADER = {
"Authorization": f"Bearer {api_key}"
}
DEFAULT_LIMIT = 100
def getRequest(endpoint, params = None, body = None, header = None):
url = BASE_URL + endpoint
return requests.get(url, params=params, json=body, headers=header)
def postRequest(endpoint, params = None, body = None, header = None):
url = BASE_URL + endpoint
return requests.post(url, params=params, json=body, headers=header)
def getCategories(strategy_id) :
params = {
"strategy_id":strategy_id
}
return getRequest("getCategories", params = params)
def getOrders(strategy_id) :
params = {
"strategy_id":strategy_id
}
return json.loads(getRequest("getOrders", params = params, header = HEADER).content)['response']
def deleteOrder(strategy_id, order_id) :
body = {
"strategy_id":strategy_id,
"order_id":order_id
}
return json.loads(postRequest("deleteOrder", body = body, header = HEADER).content)['response']
def deleteAllOrders(strategy_id) :
orders = getOrders(strategy_id)
for order in orders :
order_id = order['order_id']
deleteOrder(strategy_id, order_id)
def getPositions(strategy_id) :
params = {
"strategy_id":strategy_id
}
return json.loads(getRequest("getPositions", params = params, header=HEADER).content)['response']
def newPost(strategy_id, message = "", url = "") :
body = {
"strategy_id":strategy_id,
"description":message,
"url":url
}
return postRequest("newPost", body=body, header=HEADER)
def getPositionById(strategy_id, stock_id) :
response = getPositions(strategy_id)
for i in range(0, response.__len__()) :
if response[i]['stock_id'] == stock_id :
return response[i] #returns position data if exists
return None #returns none if position doesn't exist
def positionExists(strategy_id, stock_id) :
if getPositionById(strategy_id, stock_id) is not None :
return True
else :
return False
def getPositionBalance(strategy_id, stock_id) :
position_exists = positionExists(strategy_id, stock_id)
if position_exists:
position = getPositionById(strategy_id, stock_id)
return float(position['amount'])
else :
return 0
def buyPosition(strategy, stock_id, total) :
body = {
"strategy_id":strategy,
"stock_id":stock_id,
"type":"market",
"action":"buy",
"total":total
}
return postRequest("newOrder", body=body, header=HEADER)
def sellPosition(strategy, stock_id, amount) :
body = {
"strategy_id":strategy,
"stock_id":stock_id,
"type":"market",
"action":"sell",
"amount":amount
}
return postRequest("newOrder", body=body, header=HEADER)
def getStockHistory(stock_id, start=None, end=None, limit=DEFAULT_LIMIT) :
print(f"Getting stock data for {stock_id}")
params = {
"stock_id":stock_id,
"limit":limit
}
return getRequest("getStockPriceHistory", params = params).json()['response']