mgrotke/mgconky
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Improved comments

Browse source
This commit is contained in:
Matthew Grotke 2026-01-16 00:21:07 -05:00
parent fa2717e1f7
commit 08e22632bc
2 changed files with 56 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

36
stocks/get_stocks_alphavantage.py
View file

@ -8,12 +8,18 @@ import argparse
import requests
from datetime import datetime, timedelta
# Directory where per-symbol cache JSON files are stored
CACHE_DIR = os.path.expanduser("~/.cache/mgconky/stocks_alphavantage/")
# Delay between API calls to respect AlphaVantage rate limits
API_DELAY_SECONDS = 1
NO_THRASH_SECONDS = 60 * 60 # 1 hour
# Minimum age before re-querying a symbol to avoid API thrashing
NO_THRASH_SECONDS = 60 * 60 # 1 hour
def fetch_intraday_data(api_key, symbol, interval="1min"):
# Query AlphaVantage intraday endpoint for the given symbol
url = "https://www.alphavantage.co/query"
params = {
"function": "TIME_SERIES_INTRADAY",
@ -28,6 +34,7 @@ def fetch_intraday_data(api_key, symbol, interval="1min"):
time_series = data.get(f"Time Series ({interval})")
if time_series:
# Use earliest bar as open price and latest bar as current price
sorted_timestamps = sorted(time_series.keys())
open_time = sorted_timestamps[0]
latest_time = sorted_timestamps[-1]
@ -37,16 +44,20 @@ def fetch_intraday_data(api_key, symbol, interval="1min"):
"compare_price": float(time_series[open_time]["1. open"]),
}
else:
# API responded but did not include expected time series data
print(f"Error: No 'Time Series ({interval})' data found for {symbol}.", file=sys.stderr)
else:
# HTTP error from AlphaVantage
print(f"Error: Failed to fetch intraday data for {symbol} (HTTP {response.status_code})", file=sys.stderr)
except requests.RequestException as e:
# Network or request failure
print(f"Error: Failed to fetch intraday data for {symbol} - {e}", file=sys.stderr)
return None
def fetch_historical_data(api_key, symbol, range_in_days):
# Query AlphaVantage daily endpoint for historical comparison
url = "https://www.alphavantage.co/query"
params = {
"function": "TIME_SERIES_DAILY",
@ -60,10 +71,14 @@ def fetch_historical_data(api_key, symbol, range_in_days):
time_series = data.get("Time Series (Daily)")
if time_series:
# Most recent trading day's close is the current price
latest_date = max(time_series.keys())
current_price = float(time_series[latest_date]["4. close"])
# Target calendar date for comparison
target = datetime.now() - timedelta(days=range_in_days)
# Walk backward up to 10 days to find a valid trading day
historical_price = None
for offset in range(10):
d = (target - timedelta(days=offset)).strftime("%Y-%m-%d")
@ -76,29 +91,36 @@ def fetch_historical_data(api_key, symbol, range_in_days):
"compare_price": historical_price,
}
else:
# API responded but did not include expected daily series data
print(f"Error: No 'Time Series (Daily)' data found for {symbol}.", file=sys.stderr)
else:
# HTTP error from AlphaVantage
print(f"Error: Failed to fetch historical data for {symbol} (HTTP {response.status_code})", file=sys.stderr)
except requests.RequestException as e:
# Network or request failure
print(f"Error: Failed to fetch historical data for {symbol} - {e}", file=sys.stderr)
return None
def main():
# Parse command-line arguments
parser = argparse.ArgumentParser()
parser.add_argument("--api_key", required=True)
parser.add_argument("--symbols", required=True)
parser.add_argument("--range_in_days", type=int, default=0)
args = parser.parse_args()
# Normalize and split ticker symbols
symbols = args.symbols.strip().upper().split(",")
# Ensure cache directory exists
os.makedirs(CACHE_DIR, exist_ok=True)
for i, symbol in enumerate(symbols):
cache_path = os.path.join(CACHE_DIR, f"{symbol}.json")
# --- No-thrash protection ---
# --- No-thrash protection: skip API call if cache is still fresh ---
if os.path.exists(cache_path):
try:
with open(cache_path, "r") as f:
@ -110,18 +132,20 @@ def main():
and isinstance(payload["timestamp"], (int, float))
and (time.time() - payload["timestamp"]) < NO_THRASH_SECONDS
):
# Cache is recent enough ---> skip API call
# Cached data is recent enough -> skip querying AlphaVantage
continue
except Exception:
# Any error ---> fall through and refetch
# Any read/parse error -> ignore cache and refetch
pass
# Fetch either intraday or historical data depending on range
fetched_data = (
fetch_intraday_data(args.api_key, symbol)
if args.range_in_days < 1
else fetch_historical_data(args.api_key, symbol, args.range_in_days)
)
# Only write cache if fetched data is structurally valid
if (
isinstance(fetched_data, dict)
and "current_price" in fetched_data
@ -130,6 +154,7 @@ def main():
tmp = os.path.join(CACHE_DIR, f"{symbol}.json.tmp")
final = os.path.join(CACHE_DIR, f"{symbol}.json")
# Payload includes metadata for staleness and debugging
payload = {
"symbol": symbol,
"range_in_days": args.range_in_days,
@ -137,13 +162,16 @@ def main():
"data": fetched_data,
}
# Atomic write to avoid corrupting existing cache
with open(tmp, "w") as f:
json.dump(payload, f)
os.replace(tmp, final)
# Rate-limit delay between symbols
if i < len(symbols) - 1:
time.sleep(API_DELAY_SECONDS)
if __name__ == "__main__":
main()

35
stocks/process_stocks_alphavantage.py
View file

@ -7,10 +7,11 @@ import json
import argparse
from datetime import datetime, timedelta
# Directory containing cached per-symbol JSON files
CACHE_DIR = os.path.expanduser("~/.cache/mgconky/stocks_alphavantage/")
def main():
# Parse command-line arguments (same shape as original, minus api_key)
# Parse command-line arguments (reader does not need api_key)
parser = argparse.ArgumentParser(description="Process cached Alpha Vantage stock data.")
parser.add_argument("--symbols", required=True, help="Comma-separated list of stock symbols")
parser.add_argument("--range_in_days", type=int, default=0, help="Number of days for historical comparison")
@ -19,13 +20,13 @@ def main():
parser.add_argument("--stale_seconds", type=int, default=13 * 3600, help="Seconds before cached data is considered stale")
args = parser.parse_args()
# Split symbols (unchanged)
# Split ticker symbols by comma
symbols = args.symbols.strip().upper().split(",")
# Output builder (unchanged)
# Collect formatted output lines for Conky
output = []
# Formatting (unchanged)
# Conky color and layout formatting
color_header = "${color}"
color_label = "${color3}"
color_value = "${color3}"
@ -35,11 +36,11 @@ def main():
line_tab2_offset = "${goto 90}"
line_tab3_offset = "${alignr}"
# Iterate symbols (structure preserved)
# Process each symbol
for symbol in symbols:
cache_path = os.path.join(CACHE_DIR, f"{symbol}.json")
# Missing cache file → invalid
# Missing cache file -- display formatted placeholder
if not os.path.exists(cache_path):
output.append(
f"{line_tab1_offset}{color_bad}{symbol}: "
@ -48,19 +49,25 @@ def main():
)
continue
# Determine staleness from mtime
# Determine cache age using file modification time
age_seconds = time.time() - os.stat(cache_path).st_mtime
# Symbol turns red (color_bad) if cached data is stale
symbol_color = color_bad if age_seconds > args.stale_seconds else color_label
# Load cached JSON
# Load cached JSON payload
try:
with open(cache_path, "r") as f:
payload = json.load(f)
# Validate payload structure before use
if not isinstance(payload, dict) or "data" not in payload:
raise ValueError("Invalid cache payload")
fetched_data = payload["data"]
except Exception:
# Corrupt or unreadable cache file -- display formatted placeholder
output.append(
f"{line_tab1_offset}{color_bad}{symbol}: "
f"{line_tab2_offset}{color_value}-- "
@ -79,7 +86,7 @@ def main():
current_price = fetched_data["current_price"]
compare_price = fetched_data["compare_price"]
# Defensive: compare_price may still be None
# Comparison price may be missing if no trading day was found
if not isinstance(compare_price, (int, float)):
symbol_color = color_bad
output.append(
@ -89,15 +96,18 @@ def main():
)
continue
# Compute absolute and percentage change
price_difference = current_price - compare_price
percent_change = (price_difference / current_price) * 100
# Color change based on price movement direction
color_dynamic = (
color_good if round(price_difference, args.price_dec_places) > 0
else color_bad if round(price_difference, args.price_dec_places) < 0
else color_value
)
# The actual formatted line with valid stock data
output.append(
f"{line_tab1_offset}{symbol_color}{symbol}: "
f"{line_tab2_offset}{color_value}{round(current_price, args.price_dec_places):.{args.price_dec_places}f} "
@ -105,21 +115,24 @@ def main():
f"({round(percent_change, args.percent_dec_places):+.{args.percent_dec_places}f}%)"
)
else:
# Invalid cached data → treat as bad/stale
# Cached data missing required fields -- display formatted placeholder
output.append(
f"{line_tab1_offset}{color_bad}{symbol}: "
f"{line_tab2_offset}{color_value}-- "
f"{line_tab3_offset}{color_value}-- (--%)"
)
# Header (unchanged naming)
# Header label depends on intraday vs historical mode
header_label = "Intraday" if args.range_in_days < 1 else f"{args.range_in_days} Day"
# Formatted header line
header_line = (
f"{line_tab1_offset}{color_header}Ticker"
f"{line_tab2_offset}Price ($$)"
f"{line_tab3_offset}{header_label}{color_label}"
)
# Print final output for Conky to render
print(
header_line
+ "\n"