232 lines
10 KiB
Python
Executable File
232 lines
10 KiB
Python
Executable File
from dotenv import load_dotenv
|
|
import os
|
|
import tweepy
|
|
from requests_oauthlib import OAuth1Session
|
|
from benzinga import financial_data
|
|
from datetime import datetime, timedelta
|
|
from collections import defaultdict
|
|
import requests
|
|
import json
|
|
import ujson
|
|
import sqlite3
|
|
load_dotenv()
|
|
|
|
api_key = os.getenv('BENZINGA_API_KEY')
|
|
fin = financial_data.Benzinga(api_key)
|
|
|
|
consumer_key = os.getenv('TWITTER_API_KEY')
|
|
consumer_secret = os.getenv('TWITTER_API_SECRET')
|
|
access_token = os.getenv('TWITTER_ACCESS_TOKEN')
|
|
access_token_secret = os.getenv('TWITTER_ACCESS_TOKEN_SECRET')
|
|
|
|
con = sqlite3.connect('stocks.db')
|
|
cursor = con.cursor()
|
|
cursor.execute("PRAGMA journal_mode = wal")
|
|
cursor.execute("SELECT DISTINCT symbol FROM stocks")
|
|
stock_symbols = [row[0] for row in cursor.fetchall()]
|
|
|
|
|
|
def send_tweet(message):
|
|
# Be sure to add replace the text of the with the text you wish to Tweet. You can also add parameters to post polls, quote Tweets, Tweet with reply settings, and Tweet to Super Followers in addition to other features.
|
|
payload = {"text": message}
|
|
|
|
# Make the request
|
|
oauth = OAuth1Session(
|
|
consumer_key,
|
|
client_secret=consumer_secret,
|
|
resource_owner_key=access_token,
|
|
resource_owner_secret=access_token_secret,
|
|
)
|
|
|
|
# Making the request
|
|
response = oauth.post(
|
|
"https://api.twitter.com/2/tweets",
|
|
json=payload,
|
|
)
|
|
|
|
if response.status_code != 201:
|
|
raise Exception(
|
|
"Request returned an error: {} {}".format(response.status_code, response.text)
|
|
)
|
|
|
|
print("Response code: {}".format(response.status_code))
|
|
|
|
# Saving the response as JSON
|
|
json_response = response.json()
|
|
print(json.dumps(json_response, indent=4, sort_keys=True))
|
|
|
|
|
|
def get_news():
|
|
start_date = datetime.today().strftime('%Y-%m-%d')
|
|
end_date = start_date
|
|
url = "https://api.benzinga.com/api/v2/news"
|
|
querystring = {"token":api_key,"channels":"WIIM","dateFrom":start_date,"dateTo":end_date}
|
|
headers = {"accept": "application/json"}
|
|
response = requests.request("GET", url, headers=headers, params=querystring)
|
|
data = ujson.loads(response.text)
|
|
|
|
res_list = []
|
|
for item in data:
|
|
title = item['title']
|
|
stock_names = ' '.join(['$' + stock['name'] for stock in item['stocks']])
|
|
message = '{} {}'.format(stock_names, title)
|
|
send_tweet(message)
|
|
print(message)
|
|
|
|
def get_analyst_ratings():
|
|
url = "https://api.benzinga.com/api/v2.1/calendar/ratings"
|
|
querystring = {"token":api_key,"parameters[date_from]":"2024-04-16","parameters[date_to]":"2024-04-16"}
|
|
headers = {"accept": "application/json"}
|
|
response = requests.request("GET", url, headers=headers, params=querystring)
|
|
data = ujson.loads(response.text)['ratings']
|
|
|
|
for item in data:
|
|
symbol = item['ticker']
|
|
try:
|
|
item['adjusted_pt_current'] = round(float(item['adjusted_pt_current']))
|
|
item['adjusted_pt_prior'] = round(float(item['adjusted_pt_prior']))
|
|
except:
|
|
pass
|
|
if item['rating_current'] == 'Strong Sell' or item['rating_current'] == 'Strong Buy':
|
|
pass
|
|
elif item['rating_current'] == 'Neutral':
|
|
item['rating_current'] = 'Hold'
|
|
elif item['rating_current'] == 'Equal-Weight' or item['rating_current'] == 'Sector Weight' or item['rating_current'] == 'Sector Perform':
|
|
item['rating_current'] = 'Hold'
|
|
elif item['rating_current'] == 'In-Line':
|
|
item['rating_current'] = 'Hold'
|
|
elif item['rating_current'] == 'Outperform' and item['action_company'] == 'Downgrades':
|
|
item['rating_current'] = 'Hold'
|
|
elif item['rating_current'] == 'Negative':
|
|
item['rating_current'] = 'Sell'
|
|
elif (item['rating_current'] == 'Outperform' or item['rating_current'] == 'Overweight') and (item['action_company'] == 'Reiterates' or item['action_company'] == 'Initiates Coverage On'):
|
|
item['rating_current'] = 'Buy'
|
|
item['action_comapny'] = 'Initiates'
|
|
elif item['rating_current'] == 'Market Outperform' and (item['action_company'] == 'Maintains' or item['action_company'] == 'Reiterates'):
|
|
item['rating_current'] = 'Buy'
|
|
elif item['rating_current'] == 'Outperform' and (item['action_company'] == 'Maintains' or item['action_pt'] == 'Announces' or item['action_company'] == 'Upgrades'):
|
|
item['rating_current'] = 'Buy'
|
|
elif item['rating_current'] == 'Buy' and (item['action_company'] == 'Raises' or item['action_pt'] == 'Raises'):
|
|
item['rating_current'] = 'Strong Buy'
|
|
elif item['rating_current'] == 'Overweight' and (item['action_company'] == 'Maintains' or item['action_company'] == 'Upgrades' or item['action_company'] == 'Reiterates' or item['action_pt'] == 'Raises'):
|
|
item['rating_current'] = 'Buy'
|
|
elif item['rating_current'] == 'Positive' or item['rating_current'] == 'Sector Outperform':
|
|
item['rating_current'] = 'Buy'
|
|
elif item['rating_current'] == 'Underperform' or item['rating_current'] == 'Underweight':
|
|
item['rating_current'] = 'Sell'
|
|
elif item['rating_current'] == 'Reduce' and (item['action_company'] == 'Downgrades' or item['action_pt'] == 'Lowers'):
|
|
item['rating_current'] = 'Sell'
|
|
elif item['rating_current'] == 'Sell' and item['action_pt'] == 'Announces':
|
|
item['rating_current'] = 'Strong Sell'
|
|
elif item['rating_current'] == 'Market Perform':
|
|
item['rating_current'] = 'Hold'
|
|
elif item['rating_prior'] == 'Outperform' and item['action_company'] == 'Downgrades':
|
|
item['rating_current'] = 'Hold'
|
|
elif item['rating_current'] == 'Peer Perform' and item['rating_prior'] == 'Peer Perfrom':
|
|
item['rating_current'] = 'Hold'
|
|
elif item['rating_current'] == 'Peer Perform' and item['action_pt'] == 'Announces':
|
|
item['rating_current'] = 'Hold'
|
|
item['action_comapny'] = 'Initiates'
|
|
if symbol in stock_symbols:
|
|
message = f"{item['action_company']} {item['rating_current']} rating on ${item['ticker']} from ${item['adjusted_pt_prior']} to ${item['adjusted_pt_current']} new Price Target from {item['analyst']}."
|
|
#print(message)
|
|
send_tweet(message)
|
|
|
|
|
|
def get_analyst_insight():
|
|
url = "https://api.benzinga.com/api/v1/analyst/insights"
|
|
querystring = {"token": api_key}
|
|
response = requests.request("GET", url, params=querystring)
|
|
data = ujson.loads(response.text)['analyst-insights']
|
|
#print(data)
|
|
for item in data:
|
|
try:
|
|
symbol = item['security']['symbol']
|
|
with open(f"json/trend-analysis/{symbol}.json", 'r') as file:
|
|
ai_model = ujson.load(file)
|
|
for i in ai_model:
|
|
if i['label']== 'threeMonth':
|
|
sentiment = i['sentiment']
|
|
accuracy = i['accuracy']
|
|
if symbol in stock_symbols:
|
|
#tweet = f"{item['action']} {item['rating']} rating on ${item['security']['symbol']} with ${item['pt']} Price Target from {item['firm']}. \
|
|
#\nOur own AI Model predicts a {sentiment} Trend for the next 3 months with an accuracy of {accuracy}%."
|
|
message = f"{item['action']} {item['rating']} rating on ${item['security']['symbol']} with ${item['pt']} Price Target from {item['firm']}."
|
|
print(message)
|
|
#send_tweet(message)
|
|
except:
|
|
pass
|
|
|
|
def get_biggest_options_activity():
|
|
# Initialize dictionaries to store cumulative sums and counts
|
|
call_volume_sum = defaultdict(int)
|
|
put_volume_sum = defaultdict(int)
|
|
volume_sum = defaultdict(int)
|
|
open_interest_sum = defaultdict(int)
|
|
price_sum = defaultdict(float)
|
|
cost_basis_sum = defaultdict(float)
|
|
call_count = defaultdict(int)
|
|
put_count = defaultdict(int)
|
|
|
|
try:
|
|
end_date = datetime.today()
|
|
start_date = (end_date -timedelta(10))
|
|
start_date_str = start_date.strftime('%Y-%m-%d')
|
|
end_date_str = end_date.strftime('%Y-%m-%d')
|
|
res_list = []
|
|
for page in range(0,50):
|
|
try:
|
|
data = fin.options_activity(date_from=start_date_str, date_to=end_date_str, page=0, pagesize=1000)
|
|
data = ujson.loads(fin.output(data))['option_activity']
|
|
res_list +=data
|
|
except:
|
|
break
|
|
|
|
#filtered_data = [{key: value for key, value in item.items() if key in ['ticker','cost_basis','date_expiration','open_interest','price', 'put_call','strike_price', 'volume']} for item in res_list]
|
|
# Iterate through the data
|
|
for item in res_list:
|
|
ticker = item['ticker']
|
|
if item['put_call'] == 'CALL':
|
|
call_volume_sum[ticker] += int(item['volume'])
|
|
call_count[ticker] += 1
|
|
elif item['put_call'] == 'PUT':
|
|
put_volume_sum[ticker] += int(item['volume'])
|
|
put_count[ticker] += 1
|
|
volume_sum[ticker] += int(item['volume'])
|
|
#open_interest_sum[ticker] += int(item['open_interest'])
|
|
#price_sum[ticker] += float(item['price'])
|
|
#cost_basis_sum[ticker] += float(item['cost_basis'])
|
|
|
|
sorted_volume = sorted(volume_sum.items(), key=lambda x: x[1], reverse=True)
|
|
output = []
|
|
for i, (ticker, volume) in enumerate(sorted_volume[:3], 1):
|
|
flow_sentiment = 'Neutral'
|
|
if put_volume_sum[ticker] > call_volume_sum[ticker]:
|
|
flow_sentiment = 'Bearish'
|
|
elif put_volume_sum[ticker] < call_volume_sum[ticker]:
|
|
flow_sentiment = 'Bullish'
|
|
|
|
output.append(f"{i}) ${ticker}\n \
|
|
- Call Flow: {call_volume_sum[ticker]:,}\n \
|
|
- Put Flow: {put_volume_sum[ticker]:,}\n \
|
|
- Put/Call Ratio: {round(put_volume_sum[ticker]/call_volume_sum[ticker],2)}\n \
|
|
- Flow Sentiment: {flow_sentiment}")
|
|
|
|
message = f"Market Recap: Top 3 Highest Options Activity from this Week\n\
|
|
{output[0]}\n\
|
|
{output[1]}\n\
|
|
{output[2]}"
|
|
print(message)
|
|
#send_tweet(message)
|
|
|
|
except Exception as e:
|
|
print(e)
|
|
|
|
|
|
|
|
|
|
if __name__ == '__main__':
|
|
get_news()
|
|
#get_analyst_insight()
|
|
#get_analyst_ratings()
|
|
#get_biggest_options_activity() |