Tools/huheng-research
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: enhance watchlist workflow

Browse Source
This commit is contained in:
shay7sev
2026-06-24 14:43:58 +08:00
parent f8fb09d2a9
commit f1ad5a7d15
7 changed files with 829 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
apps/api/migrations/202606240004_watchlist_enhancements.sql Normal file
View File

@@ -0,0 +1,21 @@
ALTER TABLE app.watchlist_items
ADD COLUMN IF NOT EXISTS label TEXT NOT NULL DEFAULT 'general',
ADD COLUMN IF NOT EXISTS priority TEXT NOT NULL DEFAULT 'medium',
ADD COLUMN IF NOT EXISTS trigger_operator TEXT NOT NULL DEFAULT 'lte',
ADD COLUMN IF NOT EXISTS trigger_price_psm DOUBLE PRECISION CHECK (trigger_price_psm >= 0),
ADD COLUMN IF NOT EXISTS last_reviewed_at TIMESTAMPTZ;
CREATE TABLE IF NOT EXISTS app.watchlist_events (
event_id BIGSERIAL PRIMARY KEY,
watchlist_item_id BIGINT NOT NULL REFERENCES app.watchlist_items(watchlist_item_id) ON DELETE CASCADE,
event_type TEXT NOT NULL,
summary TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS idx_watchlist_items_label
ON app.watchlist_items(label);
CREATE INDEX IF NOT EXISTS idx_watchlist_items_priority
ON app.watchlist_items(priority);
CREATE INDEX IF NOT EXISTS idx_watchlist_events_item_created
ON app.watchlist_events(watchlist_item_id, created_at DESC);

231
apps/api/src/handlers.rs
View File

@@ -8,12 +8,13 @@ use crate::models::{
AreaComparison, AreaDetail, AreaMonthlyMetric, AreaProfile, AreaScore, AreaScoreLineage,
AreaScoreLineageQuery, AreaScoreModelRunResponse, AreaScoreSummary, CompareQuery,
ComparisonMetric, ComparisonMetricValue, CreateAreaScoreModelRun, CreateDataSource,
CreateIngestionRun, CreateRawArtifact, CreateWatchlistItem, DataSource, FinishIngestionRun,
ImportExecutionRequest, ImportExecutionResponse, ImportValidationRequest,
CreateIngestionRun, CreateRawArtifact, CreateWatchlistEvent, CreateWatchlistItem, DataSource,
FinishIngestionRun, ImportExecutionRequest, ImportExecutionResponse, ImportValidationRequest,
ImportValidationResponse, IngestionRun, IngestionRunQuery, MarketOverview, ModelRun,
MonthQuery, Neighborhood, NeighborhoodComparison, NeighborhoodComparisonItem,
NeighborhoodDetail, NeighborhoodMonthlyMetric, NeighborhoodQuery, NeighborhoodScore,
NeighborhoodScoreQuery, RawArtifact, RawArtifactQuery, UpdateWatchlistItem, WatchlistItem,
NeighborhoodScoreQuery, RawArtifact, RawArtifactQuery, UpdateWatchlistItem, WatchlistEvent,
WatchlistItem, WatchlistQuery,
};
use crate::scoring::{compute_area_scores, previous_month, AreaMetricRow, ComputedAreaScore};
use crate::state::AppState;
@@ -660,7 +661,14 @@ pub async fn get_neighborhood_detail(
area_id,
target_price_psm,
status,
notes
label,
priority,
trigger_operator,
trigger_price_psm,
notes,
created_at::text AS created_at,
updated_at::text AS updated_at,
last_reviewed_at::text AS last_reviewed_at
FROM app.watchlist_items
WHERE neighborhood_id = $1
AND status <> 'archived'
@@ -1095,6 +1103,7 @@ pub async fn create_raw_artifact(
pub async fn list_watchlist_items(
State(state): State<AppState>,
Query(query): Query<WatchlistQuery>,
) -> ApiResult<Json<Vec<WatchlistItem>>> {
let items = sqlx::query_as::<_, WatchlistItem>(
r#"
@@ -1104,12 +1113,27 @@ pub async fn list_watchlist_items(
area_id,
target_price_psm,
status,
notes
label,
priority,
trigger_operator,
trigger_price_psm,
notes,
created_at::text AS created_at,
updated_at::text AS updated_at,
last_reviewed_at::text AS last_reviewed_at
FROM app.watchlist_items
WHERE status <> 'archived'
WHERE ($1::text IS NULL OR status = $1)
AND ($1::text IS NOT NULL OR status <> 'archived')
AND ($2::text IS NULL OR area_id = $2)
AND ($3::text IS NULL OR neighborhood_id = $3)
AND ($4::text IS NULL OR label = $4)
ORDER BY updated_at DESC, watchlist_item_id DESC
"#,
)
.bind(query.status)
.bind(query.area_id)
.bind(query.neighborhood_id)
.bind(query.label)
.fetch_all(&state.pool)
.await?;
@@ -1124,30 +1148,57 @@ pub async fn create_watchlist_item(
.validate()
.map_err(|message| ApiError::BadRequest(message.to_string()))?;
let mut tx = state.pool.begin().await?;
let item = sqlx::query_as::<_, WatchlistItem>(
r#"
INSERT INTO app.watchlist_items (
neighborhood_id,
area_id,
target_price_psm,
label,
priority,
trigger_operator,
trigger_price_psm,
notes
)
VALUES ($1, $2, $3, COALESCE($4, ''))
VALUES (
$1,
$2,
$3,
COALESCE($4, 'general'),
COALESCE($5, 'medium'),
COALESCE($6, 'lte'),
$7,
COALESCE($8, '')
)
RETURNING
watchlist_item_id,
neighborhood_id,
area_id,
target_price_psm,
status,
notes
label,
priority,
trigger_operator,
trigger_price_psm,
notes,
created_at::text AS created_at,
updated_at::text AS updated_at,
last_reviewed_at::text AS last_reviewed_at
"#,
)
.bind(payload.neighborhood_id)
.bind(payload.area_id)
.bind(payload.target_price_psm)
.bind(payload.label)
.bind(payload.priority)
.bind(payload.trigger_operator)
.bind(payload.trigger_price_psm)
.bind(payload.notes)
.fetch_one(&state.pool)
.fetch_one(&mut *tx)
.await?;
insert_watchlist_event(&mut tx, item.watchlist_item_id, "created", "创建观察标的").await?;
tx.commit().await?;
Ok(Json(item))
}
@@ -1161,12 +1212,18 @@ pub async fn update_watchlist_item(
.validate()
.map_err(|message| ApiError::BadRequest(message.to_string()))?;
let mut tx = state.pool.begin().await?;
let item = sqlx::query_as::<_, WatchlistItem>(
r#"
UPDATE app.watchlist_items
SET target_price_psm = COALESCE($2, target_price_psm),
status = COALESCE($3, status),
notes = COALESCE($4, notes),
label = COALESCE($4, label),
priority = COALESCE($5, priority),
trigger_operator = COALESCE($6, trigger_operator),
trigger_price_psm = COALESCE($7, trigger_price_psm),
notes = COALESCE($8, notes),
last_reviewed_at = CASE WHEN $9 THEN now() ELSE last_reviewed_at END,
updated_at = now()
WHERE watchlist_item_id = $1
RETURNING
@@ -1175,16 +1232,44 @@ pub async fn update_watchlist_item(
area_id,
target_price_psm,
status,
notes
label,
priority,
trigger_operator,
trigger_price_psm,
notes,
created_at::text AS created_at,
updated_at::text AS updated_at,
last_reviewed_at::text AS last_reviewed_at
"#,
)
.bind(watchlist_item_id)
.bind(payload.target_price_psm)
.bind(payload.status)
.bind(payload.status.clone())
.bind(payload.label)
.bind(payload.priority)
.bind(payload.trigger_operator)
.bind(payload.trigger_price_psm)
.bind(payload.notes)
.fetch_optional(&state.pool)
.bind(payload.mark_reviewed.unwrap_or(false))
.fetch_optional(&mut *tx)
.await?
.ok_or_else(|| ApiError::BadRequest("watchlist item not found".to_string()))?;
insert_watchlist_event(
&mut tx,
item.watchlist_item_id,
if payload.mark_reviewed.unwrap_or(false) {
"reviewed"
} else {
"updated"
},
if payload.mark_reviewed.unwrap_or(false) {
"完成一次人工复核"
} else {
"更新观察标的"
},
)
.await?;
tx.commit().await?;
Ok(Json(item))
}
@@ -1193,6 +1278,7 @@ pub async fn archive_watchlist_item(
State(state): State<AppState>,
Path(watchlist_item_id): Path<i64>,
) -> ApiResult<Json<WatchlistItem>> {
let mut tx = state.pool.begin().await?;
let item = sqlx::query_as::<_, WatchlistItem>(
r#"
UPDATE app.watchlist_items
@@ -1205,17 +1291,109 @@ pub async fn archive_watchlist_item(
area_id,
target_price_psm,
status,
notes
label,
priority,
trigger_operator,
trigger_price_psm,
notes,
created_at::text AS created_at,
updated_at::text AS updated_at,
last_reviewed_at::text AS last_reviewed_at
"#,
)
.bind(watchlist_item_id)
.fetch_optional(&state.pool)
.fetch_optional(&mut *tx)
.await?
.ok_or_else(|| ApiError::BadRequest("watchlist item not found".to_string()))?;
insert_watchlist_event(&mut tx, item.watchlist_item_id, "archived", "归档观察标的").await?;
tx.commit().await?;
Ok(Json(item))
}
pub async fn list_watchlist_events(
State(state): State<AppState>,
Path(watchlist_item_id): Path<i64>,
) -> ApiResult<Json<Vec<WatchlistEvent>>> {
let events = sqlx::query_as::<_, WatchlistEvent>(
r#"
SELECT
event_id,
watchlist_item_id,
event_type,
summary,
created_at::text AS created_at
FROM app.watchlist_events
WHERE watchlist_item_id = $1
ORDER BY created_at DESC, event_id DESC
LIMIT 50
"#,
)
.bind(watchlist_item_id)
.fetch_all(&state.pool)
.await?;
Ok(Json(events))
}
pub async fn create_watchlist_event(
State(state): State<AppState>,
Path(watchlist_item_id): Path<i64>,
Json(payload): Json<CreateWatchlistEvent>,
) -> ApiResult<Json<WatchlistEvent>> {
payload
.validate()
.map_err(|message| ApiError::BadRequest(message.to_string()))?;
let event = sqlx::query_as::<_, WatchlistEvent>(
r#"
INSERT INTO app.watchlist_events (
watchlist_item_id,
event_type,
summary
)
VALUES ($1, $2, $3)
RETURNING
event_id,
watchlist_item_id,
event_type,
summary,
created_at::text AS created_at
"#,
)
.bind(watchlist_item_id)
.bind(payload.event_type.trim().to_string())
.bind(payload.summary.trim().to_string())
.fetch_one(&state.pool)
.await?;
Ok(Json(event))
}
async fn insert_watchlist_event(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
watchlist_item_id: i64,
event_type: &str,
summary: &str,
) -> Result<(), sqlx::Error> {
sqlx::query(
r#"
INSERT INTO app.watchlist_events (
watchlist_item_id,
event_type,
summary
)
VALUES ($1, $2, $3)
"#,
)
.bind(watchlist_item_id)
.bind(event_type)
.bind(summary)
.execute(&mut **tx)
.await?;
Ok(())
}
async fn fetch_area_scores(state: &AppState, month: &str) -> Result<Vec<AreaScore>, sqlx::Error> {
sqlx::query_as::<_, AreaScore>(
r#"
@@ -1788,6 +1966,10 @@ mod tests {
neighborhood_id: None,
area_id: None,
target_price_psm: None,
label: None,
priority: None,
trigger_operator: None,
trigger_price_psm: None,
notes: None,
}
.validate()
@@ -1797,6 +1979,10 @@ mod tests {
neighborhood_id: None,
area_id: Some("zhangjiang".to_string()),
target_price_psm: Some(80_000.0),
label: Some("核心".to_string()),
priority: Some("high".to_string()),
trigger_operator: Some("lte".to_string()),
trigger_price_psm: Some(76_000.0),
notes: None,
}
.validate()
@@ -1808,7 +1994,12 @@ mod tests {
assert!(UpdateWatchlistItem {
target_price_psm: Some(-1.0),
status: None,
label: None,
priority: None,
trigger_operator: None,
trigger_price_psm: None,
notes: None,
mark_reviewed: None,
}
.validate()
.is_err());
@@ -1816,7 +2007,12 @@ mod tests {
assert!(UpdateWatchlistItem {
target_price_psm: None,
status: Some("unknown".to_string()),
label: None,
priority: None,
trigger_operator: None,
trigger_price_psm: None,
notes: None,
mark_reviewed: None,
}
.validate()
.is_err());
@@ -1824,7 +2020,12 @@ mod tests {
assert!(UpdateWatchlistItem {
target_price_psm: Some(80_000.0),
status: Some("active".to_string()),
label: Some("核心".to_string()),
priority: Some("high".to_string()),
trigger_operator: Some("lte".to_string()),
trigger_price_psm: Some(76_000.0),
notes: Some("跟踪产业兑现和挂牌变化".to_string()),
mark_reviewed: Some(true),
}
.validate()
.is_ok());

117
apps/api/src/models.rs
View File

@@ -43,6 +43,14 @@ pub struct RawArtifactQuery {
pub sha256: Option<String>,
}
#[derive(Debug, Deserialize)]
pub struct WatchlistQuery {
pub status: Option<String>,
pub area_id: Option<String>,
pub neighborhood_id: Option<String>,
pub label: Option<String>,
}
#[derive(Debug, Deserialize)]
pub struct ImportValidationRequest {
pub template_name: String,
@@ -551,7 +559,14 @@ pub struct WatchlistItem {
pub area_id: Option<String>,
pub target_price_psm: Option<f64>,
pub status: String,
pub label: String,
pub priority: String,
pub trigger_operator: String,
pub trigger_price_psm: Option<f64>,
pub notes: String,
pub created_at: String,
pub updated_at: String,
pub last_reviewed_at: Option<String>,
}
#[derive(Debug, Deserialize)]
@@ -559,6 +574,10 @@ pub struct CreateWatchlistItem {
pub neighborhood_id: Option<String>,
pub area_id: Option<String>,
pub target_price_psm: Option<f64>,
pub label: Option<String>,
pub priority: Option<String>,
pub trigger_operator: Option<String>,
pub trigger_price_psm: Option<f64>,
pub notes: Option<String>,
}
@@ -570,6 +589,32 @@ impl CreateWatchlistItem {
if matches!(self.target_price_psm, Some(value) if value < 0.0) {
return Err("target_price_psm must be non-negative");
}
if matches!(self.trigger_price_psm, Some(value) if value < 0.0) {
return Err("trigger_price_psm must be non-negative");
}
if let Some(priority) = &self.priority {
if !is_watchlist_priority(priority) {
return Err("priority must be low, medium, or high");
}
}
if let Some(operator) = &self.trigger_operator {
if !is_watchlist_trigger_operator(operator) {
return Err("trigger_operator must be lte or gte");
}
}
for value in [
self.label.as_deref(),
self.priority.as_deref(),
self.trigger_operator.as_deref(),
self.notes.as_deref(),
]
.into_iter()
.flatten()
{
if has_sensitive_text(value) {
return Err("payload contains sensitive text");
}
}
Ok(())
}
}
@@ -600,7 +645,12 @@ fn is_valid_sha256(value: &str) -> bool {
pub struct UpdateWatchlistItem {
pub target_price_psm: Option<f64>,
pub status: Option<String>,
pub label: Option<String>,
pub priority: Option<String>,
pub trigger_operator: Option<String>,
pub trigger_price_psm: Option<f64>,
pub notes: Option<String>,
pub mark_reviewed: Option<bool>,
}
impl UpdateWatchlistItem {
@@ -608,11 +658,78 @@ impl UpdateWatchlistItem {
if matches!(self.target_price_psm, Some(value) if value < 0.0) {
return Err("target_price_psm must be non-negative");
}
if matches!(self.trigger_price_psm, Some(value) if value < 0.0) {
return Err("trigger_price_psm must be non-negative");
}
if let Some(status) = &self.status {
if !matches!(status.as_str(), "active" | "paused" | "archived") {
return Err("status must be active, paused, or archived");
}
}
if let Some(priority) = &self.priority {
if !is_watchlist_priority(priority) {
return Err("priority must be low, medium, or high");
}
}
if let Some(operator) = &self.trigger_operator {
if !is_watchlist_trigger_operator(operator) {
return Err("trigger_operator must be lte or gte");
}
}
for value in [
self.status.as_deref(),
self.label.as_deref(),
self.priority.as_deref(),
self.trigger_operator.as_deref(),
self.notes.as_deref(),
]
.into_iter()
.flatten()
{
if has_sensitive_text(value) {
return Err("payload contains sensitive text");
}
}
Ok(())
}
}
#[derive(Debug, Clone, Serialize, FromRow)]
pub struct WatchlistEvent {
pub event_id: i64,
pub watchlist_item_id: i64,
pub event_type: String,
pub summary: String,
pub created_at: String,
}
#[derive(Debug, Deserialize)]
pub struct CreateWatchlistEvent {
pub event_type: String,
pub summary: String,
}
impl CreateWatchlistEvent {
pub fn validate(&self) -> Result<(), &'static str> {
if self.event_type.trim().is_empty() {
return Err("event_type is required");
}
if self.summary.trim().is_empty() {
return Err("summary is required");
}
for value in [self.event_type.as_str(), self.summary.as_str()] {
if has_sensitive_text(value) {
return Err("payload contains sensitive text");
}
}
Ok(())
}
}
fn is_watchlist_priority(priority: &str) -> bool {
matches!(priority, "low" | "medium" | "high")
}
fn is_watchlist_trigger_operator(operator: &str) -> bool {
matches!(operator, "lte" | "gte")
}

15
apps/api/src/routes.rs
View File

@@ -5,11 +5,12 @@ use tower_http::trace::TraceLayer;
use crate::handlers::{
archive_watchlist_item, compare_areas, compare_neighborhoods, create_area_score_model_run,
create_data_source, create_ingestion_run, create_raw_artifact, create_watchlist_item,
execute_import, finish_ingestion_run, get_area_detail, get_area_score_lineage,
get_neighborhood, get_neighborhood_detail, health, list_area_scores, list_data_sources,
list_ingestion_runs, list_neighborhood_scores, list_neighborhoods, list_raw_artifacts,
list_watchlist_items, market_overview, ready, update_watchlist_item, validate_import,
create_data_source, create_ingestion_run, create_raw_artifact, create_watchlist_event,
create_watchlist_item, execute_import, finish_ingestion_run, get_area_detail,
get_area_score_lineage, get_neighborhood, get_neighborhood_detail, health, list_area_scores,
list_data_sources, list_ingestion_runs, list_neighborhood_scores, list_neighborhoods,
list_raw_artifacts, list_watchlist_events, list_watchlist_items, market_overview, ready,
update_watchlist_item, validate_import,
};
use crate::state::AppState;
@@ -65,6 +66,10 @@ pub fn build_router(state: AppState) -> Router {
.route(
"/watchlist/{watchlist_item_id}",
patch(update_watchlist_item).delete(archive_watchlist_item),
)
.route(
"/watchlist/{watchlist_item_id}/events",
get(list_watchlist_events).post(create_watchlist_event),
),
)
.layer(TraceLayer::new_for_http())