From abb3951e0d5b2aa8b192887e31379390b7e93d8e Mon Sep 17 00:00:00 2001 From: MuslemRahimi Date: Tue, 4 Feb 2025 17:23:14 +0100 Subject: [PATCH] update unusual activity --- app/cron_unusual_activity.py | 143 +++++++++++++++-------------------- 1 file changed, 59 insertions(+), 84 deletions(-) diff --git a/app/cron_unusual_activity.py b/app/cron_unusual_activity.py index 224f544..614d1bf 100644 --- a/app/cron_unusual_activity.py +++ b/app/cron_unusual_activity.py @@ -1,8 +1,5 @@ -from __future__ import print_function import asyncio import time -import intrinio_sdk as intrinio -from intrinio_sdk.rest import ApiException from datetime import datetime, timedelta import ast import orjson @@ -11,27 +8,8 @@ import aiohttp from concurrent.futures import ThreadPoolExecutor import sqlite3 import re - -from dotenv import load_dotenv import os -load_dotenv() - -api_key = os.getenv('INTRINIO_API_KEY') - -intrinio.ApiClient().set_api_key(api_key) -#intrinio.ApiClient().allow_retries(True) - -today = datetime.today() -start_date = (today - timedelta(150)).strftime("%Y-%m-%d") -end_date = (today + timedelta(30)).strftime("%Y-%m-%d") - -next_page = '' -page_size = 1000 -activity_type = '' -sentiment = '' -minimum_total_value = 1E6 -maximum_total_value = 2E10 # Database connection and symbol retrieval @@ -51,21 +29,6 @@ def get_total_symbols(): return stocks_symbols + etf_symbols -def get_tickers_from_directory(): - directory = "json/options-historical-data/companies" - try: - # Ensure the directory exists - if not os.path.exists(directory): - raise FileNotFoundError(f"The directory '{directory}' does not exist.") - - # Get all tickers from filenames - return [file.replace(".json", "") for file in os.listdir(directory) if file.endswith(".json")] - - except Exception as e: - print(f"An error occurred: {e}") - return [] - - async def save_json(data, symbol): directory = "json/unusual-activity" os.makedirs(directory, exist_ok=True) @@ -73,68 +36,80 @@ async def save_json(data, symbol): file.write(orjson.dumps(data)) -def parse_option_symbol(option_symbol): - # Define regex pattern to match the symbol structure - match = re.match(r"([A-Z]+)(\d{6})([CP])(\d+)", option_symbol) - if not match: - raise ValueError(f"Invalid option_symbol format: {option_symbol}") - - ticker, expiration, option_type, strike_price = match.groups() - - # Convert expiration to datetime - date_expiration = datetime.strptime(expiration, "%y%m%d").date() - - # Convert strike price to float - strike_price = int(strike_price) / 1000 +async def get_dataset(): + today = datetime.today() + start_date = today - timedelta(days=365) + date_list = [(start_date + timedelta(days=i)).strftime("%Y-%m-%d") for i in range(365)] - return date_expiration, option_type, strike_price + unique_data = {} + + for date in tqdm(date_list): + try: + with open(f"json/options-historical-data/flow-data/{date}.json", "r") as file: + data = orjson.loads(file.read()) + data = [item for item in data if item['cost_basis'] >=1E6] + for item in data: + if "id" in item: + unique_data[item["id"]] = item # Store unique items based on "id" + except: + pass + + try: + with open(f"json/options-flow/feed/data.json", "r") as file: + data = orjson.loads(file.read()) + data = [item for item in data if item['cost_basis'] >=1E6] + for item in data: + if "id" in item: + unique_data[item["id"]] = item + except: + pass + + all_data = list(unique_data.values()) # Convert back to a list + + return all_data + + +async def get_data(symbol, data): -async def get_data(symbol): - response = intrinio.OptionsApi().get_unusual_activity_intraday(symbol, next_page=next_page, page_size=page_size, activity_type=activity_type, sentiment=sentiment, start_date=start_date, end_date=end_date, minimum_total_value=minimum_total_value, maximum_total_value=maximum_total_value) - data = (response.__dict__['_trades']) res_list = [] if len(data) > 0: for item in data: try: - trade_data = item.__dict__ - trade_data = {key.lstrip('_'): value for key, value in trade_data.items()} - option_symbol = trade_data['contract'].replace("___","").replace("__","").replace("_","") - date_expiration, option_type, strike_price = parse_option_symbol(option_symbol) - if trade_data['underlying_price_at_execution'] > 0: #and date_expiration >= datetime.today().date(): - - res_list.append({'date': trade_data['timestamp'].strftime("%Y-%m-%d"), - 'askprice': trade_data['ask_at_execution'], - 'bidPrice': trade_data['bid_at_execution'], - 'premium': trade_data['total_value'], - 'sentiment': trade_data['sentiment'].capitalize(), - 'avgPrice': trade_data['average_price'], - 'price': trade_data['underlying_price_at_execution'], - 'unusualType': trade_data['type'].capitalize(), - 'size': trade_data['total_size'], - 'optionSymbol': option_symbol, - 'strike': strike_price, - 'expiry': date_expiration.strftime("%Y-%m-%d"), - 'optionType': option_type.replace("P","Put").replace("C","Call") + if item['ticker'] == symbol: + res_list.append({ + 'date': item['date'], + 'premium': item['cost_basis'], + 'sentiment': item['sentiment'], + 'executionEst': item['execution_estimate'], + 'price': item['underlying_price'], + 'unusualType': item['option_activity_type'], + 'size': item['size'], + 'oi': item['open_interest'], + 'optionSymbol': item['option_symbol'], + 'strike': item['strike_price'], + 'expiry': item['date_expiration'], + 'optionType': item['put_call'], }) - except: - pass + except Exception as e: + print(e) + + res_list = sorted(res_list, key=lambda x: x['date'], reverse=True) + + if res_list: + await save_json(res_list, symbol) - res_list = sorted(res_list, key=lambda x: x['date'], reverse=True) - if res_list: - await save_json(res_list, symbol) async def main(): - total_symbols = get_tickers_from_directory() - if len(total_symbols) < 3000: - total_symbols = get_total_symbols() - print(f"Number of tickers: {len(total_symbols)}") + total_symbols = get_total_symbols() + data = await get_dataset() for symbol in tqdm(total_symbols): try: - data = await get_data(symbol) + await get_data(symbol, data) + except Exception as e: print(f"Error processing {symbol}: {e}") - + if __name__ == "__main__": asyncio.run(main())