lyman/Fundamental_Analysis
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(alphavantage): resolve 404 profile issue by serializing requests and adding rate limit protection

Browse Source
This commit is contained in:
Lv, Qi 2025-11-22 00:12:42 +08:00
parent 0cb31e363e
commit 6c880f51dd
1 changed files with 38 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
services/alphavantage-provider-service/src/worker.rs
View File

@ -136,26 +136,32 @@ async fn handle_fetch_command_inner(
// Add datatype=json to force JSON response if supported (or at least Python-dict like)
let params_quote = vec![("symbol", av_symbol.as_str()), ("datatype", "json")];
let overview_task = client.query("COMPANY_OVERVIEW", &params_overview);
let income_task = client.query("INCOME_STATEMENT", &params_income);
let balance_task = client.query("BALANCE_SHEET", &params_balance);
let cashflow_task = client.query("CASH_FLOW", &params_cashflow);
let quote_task = client.query("GLOBAL_QUOTE", &params_quote);
let overview_json = client.query("COMPANY_OVERVIEW", &params_overview).await?;
check_av_response(&overview_json)?;
tokio::time::sleep(std::time::Duration::from_secs(2)).await; // Rate limit protection
let data = match tokio::try_join!(
overview_task,
income_task,
balance_task,
cashflow_task,
quote_task
) {
Ok(data) => data,
Err(e) => {
let error_msg = format!("Failed to fetch data from AlphaVantage: {}", e);
error!(error_msg);
return Err(e);
}
};
let quote_json = client.query("GLOBAL_QUOTE", &params_quote).await?;
check_av_response(&quote_json)?;
tokio::time::sleep(std::time::Duration::from_secs(2)).await;
let income_json = client.query("INCOME_STATEMENT", &params_income).await?;
check_av_response(&income_json)?;
tokio::time::sleep(std::time::Duration::from_secs(2)).await;
let balance_json = client.query("BALANCE_SHEET", &params_balance).await?;
check_av_response(&balance_json)?;
tokio::time::sleep(std::time::Duration::from_secs(2)).await;
let cashflow_json = client.query("CASH_FLOW", &params_cashflow).await?;
check_av_response(&cashflow_json)?;
let data = (
overview_json,
income_json,
balance_json,
cashflow_json,
quote_json
);
// Write to Cache
let payload = serde_json::json!(data);
@ -203,6 +209,9 @@ async fn handle_fetch_command_inner(
warn!("Failed to parse CompanyProfile: {}", e);
}
}
} else {
// If Symbol is missing but check_av_response passed, it might be an empty object {}
warn!("COMPANY_OVERVIEW returned JSON without 'Symbol' field: {:?}", overview_json);
}
// 2.2 Financials
@ -328,6 +337,16 @@ async fn handle_fetch_command_inner(
Ok(())
}
fn check_av_response(v: &Value) -> Result<()> {
if let Some(note) = v.get("Note").and_then(|s| s.as_str()) {
return Err(AppError::Internal(format!("AlphaVantage Rate Limit: {}", note)));
}
if let Some(info) = v.get("Information").and_then(|s| s.as_str()) {
return Err(AppError::Internal(format!("AlphaVantage Information: {}", info)));
}
Ok(())
}
async fn update_task_progress(tasks: &TaskStore, request_id: Uuid, percent: u8, details: &str, status: Option<TaskStatus>) {
if let Some(mut task) = tasks.get_mut(&request_id) {
task.progress_percent = percent;