284 lines
10 KiB
Python
Executable File
284 lines
10 KiB
Python
Executable File
import aiohttp
|
|
import asyncio
|
|
import sqlite3
|
|
import certifi
|
|
import json
|
|
import pandas as pd
|
|
from tqdm import tqdm
|
|
import re
|
|
import pandas as pd
|
|
from datetime import datetime
|
|
import subprocess
|
|
import time
|
|
import warnings
|
|
from dotenv import load_dotenv
|
|
import os
|
|
|
|
# Filter out the specific RuntimeWarning
|
|
warnings.filterwarnings("ignore", category=RuntimeWarning, message="invalid value encountered in scalar divide")
|
|
|
|
con = sqlite3.connect('stocks.db')
|
|
etf_con = sqlite3.connect('etf.db')
|
|
crypto_con = sqlite3.connect('crypto.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()]
|
|
|
|
etf_cursor = etf_con.cursor()
|
|
etf_cursor.execute("PRAGMA journal_mode = wal")
|
|
etf_cursor.execute("SELECT DISTINCT symbol FROM etfs")
|
|
etf_symbols = [row[0] for row in etf_cursor.fetchall()]
|
|
|
|
crypto_cursor = crypto_con.cursor()
|
|
crypto_cursor.execute("PRAGMA journal_mode = wal")
|
|
crypto_cursor.execute("SELECT DISTINCT symbol FROM cryptos")
|
|
crypto_symbols = [row[0] for row in crypto_cursor.fetchall()]
|
|
|
|
total_symbols = stock_symbols + etf_symbols + crypto_symbols
|
|
con.close()
|
|
etf_con.close()
|
|
crypto_con.close()
|
|
|
|
|
|
load_dotenv()
|
|
api_key = os.getenv('FMP_API_KEY')
|
|
quarter_date = '2024-3-31'
|
|
|
|
|
|
if os.path.exists("backup_db/institute.db"):
|
|
os.remove('backup_db/institute.db')
|
|
|
|
|
|
def get_jsonparsed_data(data):
|
|
try:
|
|
return json.loads(data)
|
|
except json.JSONDecodeError:
|
|
return {}
|
|
|
|
|
|
class InstituteDatabase:
|
|
def __init__(self, db_path):
|
|
self.db_path = db_path
|
|
self.conn = sqlite3.connect(db_path)
|
|
self.cursor = self.conn.cursor()
|
|
self.cursor.execute("PRAGMA journal_mode = wal")
|
|
self.conn.commit()
|
|
self._create_table()
|
|
|
|
def close_connection(self):
|
|
self.cursor.close()
|
|
self.conn.close()
|
|
|
|
def _create_table(self):
|
|
self.cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS institutes (
|
|
cik TEXT PRIMARY KEY,
|
|
name TEXT
|
|
)
|
|
""")
|
|
|
|
|
|
def get_column_type(self, value):
|
|
column_type = ""
|
|
|
|
if isinstance(value, str):
|
|
column_type = "TEXT"
|
|
elif isinstance(value, int):
|
|
column_type = "INTEGER"
|
|
elif isinstance(value, float):
|
|
column_type = "REAL"
|
|
else:
|
|
# Handle other data types or customize based on your specific needs
|
|
column_type = "TEXT"
|
|
|
|
return column_type
|
|
|
|
def remove_null(self, value):
|
|
if isinstance(value, str) and value == None:
|
|
value = 'n/a'
|
|
elif isinstance(value, int) and value == None:
|
|
value = 0
|
|
elif isinstance(value, float) and value == None:
|
|
value = 0
|
|
else:
|
|
# Handle other data types or customize based on your specific needs
|
|
pass
|
|
|
|
return value
|
|
|
|
|
|
|
|
|
|
|
|
|
|
async def save_portfolio_data(self, session, cik):
|
|
try:
|
|
urls = [
|
|
f"https://financialmodelingprep.com/api/v4/institutional-ownership/industry/portfolio-holdings-summary?cik={cik}&date={quarter_date}&page=0&apikey={api_key}",
|
|
f"https://financialmodelingprep.com/api/v4/institutional-ownership/portfolio-holdings?cik={cik}&date={quarter_date}&page=0&apikey={api_key}",
|
|
f"https://financialmodelingprep.com/api/v4/institutional-ownership/portfolio-holdings-summary?cik={cik}&date={quarter_date}&page=0&apikey={api_key}"
|
|
]
|
|
|
|
portfolio_data = {}
|
|
|
|
for url in urls:
|
|
async with session.get(url) as response:
|
|
data = await response.text()
|
|
parsed_data = get_jsonparsed_data(data)
|
|
|
|
try:
|
|
if isinstance(parsed_data, list) and "industry/portfolio-holdings-summary" in url:
|
|
# Handle list response, save as JSON object
|
|
portfolio_data['industry'] = json.dumps(parsed_data)
|
|
if isinstance(parsed_data, list) and "https://financialmodelingprep.com/api/v4/institutional-ownership/portfolio-holdings?cik=" in url:
|
|
# Handle list response, save as JSON object
|
|
|
|
parsed_data = [
|
|
{**item, 'type': ('stocks' if item['symbol'] in stock_symbols else
|
|
'crypto' if item['symbol'] in crypto_symbols else
|
|
'etf' if item['symbol'] in etf_symbols else None)}
|
|
for item in parsed_data
|
|
if 'symbol' in item and item['symbol'] is not None and item['symbol'] in total_symbols
|
|
]
|
|
|
|
portfolio_data['holdings'] = json.dumps(parsed_data)
|
|
|
|
|
|
number_of_stocks = len(parsed_data)
|
|
#total_market_value = sum(item['marketValue'] for item in parsed_data)
|
|
#avg_performance_percentage = sum(item['performancePercentage'] for item in parsed_data) / len(parsed_data)
|
|
performance_percentages = [item.get("performancePercentage", 0) for item in parsed_data]
|
|
|
|
|
|
positive_performance_count = sum(1 for percentage in performance_percentages if percentage > 0)
|
|
win_rate = round(positive_performance_count / len(performance_percentages) * 100,2)
|
|
data_dict = {
|
|
'winRate': win_rate,
|
|
'numberOfStocks': number_of_stocks,
|
|
#'marketValue': total_market_value,
|
|
}
|
|
|
|
portfolio_data.update(data_dict)
|
|
|
|
elif isinstance(parsed_data, list) and "https://financialmodelingprep.com/api/v4/institutional-ownership/portfolio-holdings-summary" in url:
|
|
# Handle list response, save as JSON object
|
|
portfolio_data['summary'] = json.dumps(parsed_data)
|
|
data_dict = {
|
|
#'numberOfStocks': parsed_data[0]['portfolioSize'],
|
|
'marketValue': parsed_data[0]['marketValue'],
|
|
'averageHoldingPeriod': parsed_data[0]['averageHoldingPeriod'],
|
|
'turnover': parsed_data[0]['turnover'],
|
|
'performancePercentage3year': parsed_data[0]['performancePercentage3year'],
|
|
#'performancePercentage': parsed_data[0]['performancePercentage']
|
|
}
|
|
portfolio_data.update(data_dict)
|
|
|
|
|
|
|
|
except:
|
|
pass
|
|
|
|
# Check if columns already exist in the table
|
|
self.cursor.execute("PRAGMA table_info(institutes)")
|
|
columns = {column[1]: column[2] for column in self.cursor.fetchall()}
|
|
|
|
holdings_list = json.loads(portfolio_data['holdings'])
|
|
|
|
symbols_to_check = {holding['symbol'] for holding in holdings_list[:3]} # Extract the first two symbols
|
|
symbols_not_in_list = not any(symbol in total_symbols for symbol in symbols_to_check)
|
|
|
|
|
|
if symbols_not_in_list or 'industry' not in portfolio_data or len(json.loads(portfolio_data['industry'])) == 0:
|
|
# If 'industry' is not a list, delete the row and return
|
|
#print(f"Deleting row for cik {cik} because 'industry' is not a list.")
|
|
self.cursor.execute("DELETE FROM institutes WHERE cik = ?", (cik,))
|
|
self.conn.commit()
|
|
return
|
|
|
|
# Update column definitions with keys from portfolio_data
|
|
column_definitions = {
|
|
key: (self.get_column_type(portfolio_data.get(key, None)), self.remove_null(portfolio_data.get(key, None)))
|
|
for key in portfolio_data
|
|
}
|
|
|
|
for column, (column_type, value) in column_definitions.items():
|
|
if column not in columns and column_type:
|
|
self.cursor.execute(f"ALTER TABLE institutes ADD COLUMN {column} {column_type}")
|
|
|
|
self.cursor.execute(f"UPDATE institutes SET {column} = ? WHERE cik = ?", (value, cik))
|
|
|
|
self.conn.commit()
|
|
|
|
except Exception as e:
|
|
print(f"Failed to fetch portfolio data for cik {cik}: {str(e)}")
|
|
|
|
|
|
|
|
async def save_insitute(self, institutes):
|
|
|
|
institute_data = []
|
|
|
|
for item in institutes:
|
|
cik = item.get('cik', '')
|
|
name = item.get('name', '')
|
|
|
|
|
|
institute_data.append((cik, name))
|
|
|
|
|
|
self.cursor.execute("BEGIN TRANSACTION") # Begin a transaction
|
|
|
|
for data in institute_data:
|
|
cik, name = data
|
|
self.cursor.execute("""
|
|
INSERT OR IGNORE INTO institutes (cik, name)
|
|
VALUES (?, ?)
|
|
""", (cik, name))
|
|
self.cursor.execute("""
|
|
UPDATE institutes SET name = ?
|
|
WHERE cik = ?
|
|
""", (name, cik))
|
|
|
|
self.cursor.execute("COMMIT") # Commit the transaction
|
|
self.conn.commit()
|
|
|
|
|
|
|
|
# Save OHLC data for each ticker using aiohttp
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
tasks = []
|
|
i = 0
|
|
for item in tqdm(institute_data):
|
|
cik, name = item
|
|
tasks.append(self.save_portfolio_data(session, cik))
|
|
|
|
i += 1
|
|
if i % 700 == 0:
|
|
await asyncio.gather(*tasks)
|
|
tasks = []
|
|
print('sleeping mode: ', i)
|
|
await asyncio.sleep(60) # Pause for 60 seconds
|
|
|
|
|
|
if tasks:
|
|
await asyncio.gather(*tasks)
|
|
|
|
|
|
url = f"https://financialmodelingprep.com/api/v4/institutional-ownership/list?apikey={api_key}"
|
|
|
|
|
|
async def fetch_tickers():
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(url) as response:
|
|
data = await response.text()
|
|
return get_jsonparsed_data(data)
|
|
|
|
|
|
db = InstituteDatabase('backup_db/institute.db')
|
|
loop = asyncio.get_event_loop()
|
|
all_tickers = loop.run_until_complete(fetch_tickers())
|
|
loop.run_until_complete(db.save_insitute(all_tickers))
|
|
db.close_connection() |