在Neo4j中使用Java插件运行情感分析

mkdir -p ~/neo4j/plugins ~/neo4j/data
docker run \
  --name neo4j-sentiment \
  -p 7474:7474 -p 7687:7687 \
  -v ~/neo4j/plugins:/plugins \
  -v ~/neo4j/data:/data \
  -e NEO4J_AUTH=neo4j/password \
  -e NEO4J_dbms_security_procedures_allowlist="sentiment.*" \
  neo4j:2026.06.0

mkdir neo4j-sentiment-udf
cd neo4j-sentiment-udf

neo4j-sentiment-udf/
├── pom.xml
└── src/
    ├── main/
    │   └── java/
    │       └── sentiment/
    │           └── Sentimentable.java
    └── test/
        └── java/
            └── sentiment/
                └── SentimentableTest.java

mkdir -p src/main/java/sentiment
mkdir -p src/test/java/sentiment

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>org.neo4j.example</groupId>
  <artifactId>sentimentable</artifactId>
  <version>1.0.0-SNAPSHOT</version>
  <packaging>jar</packaging>
  <name>Neo4j Sentiment UDF</name>
  <description>VADER sentiment analysis as a Neo4j user-defined function</description>

  <!-- 调整 1：使用本文所用的 Java 21 和 Neo4j 2026.06.0 -->
  <properties>
    <java.version>21</java.version>
    <neo4j.version>2026.06.0</neo4j.version>
  </properties>

  <!-- 调整 2：为 VaderSentimentJava 添加 JitPack 仓库 -->
  <repositories>
    <repository>
      <id>jitpack.io</id>
      <url>https://jitpack.io</url>
    </repository>
  </repositories>

  <dependencies>
    <dependency>
      <groupId>org.neo4j</groupId>
      <artifactId>neo4j</artifactId>
      <version>${neo4j.version}</version>
      <scope>provided</scope>
    </dependency>
    <dependency>
      <groupId>com.github.apanimesh061</groupId>
      <artifactId>VaderSentimentJava</artifactId>
      <version>v1.1.1</version>
    </dependency>
    <dependency>
      <groupId>org.neo4j.test</groupId>
      <artifactId>neo4j-harness</artifactId>
      <version>${neo4j.version}</version>
      <scope>test</scope>
    </dependency>
    <dependency>
      <groupId>org.neo4j.driver</groupId>
      <artifactId>neo4j-java-driver</artifactId>
      <version>6.0.2</version>
      <scope>test</scope>
    </dependency>
  </dependencies>

  <build>
    <plugins>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-compiler-plugin</artifactId>
        <version>3.11.0</version>
        <configuration>
          <source>21</source>
          <target>21</target>
        </configuration>
      </plugin>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-surefire-plugin</artifactId>
        <version>3.5.4</version>
      </plugin>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-shade-plugin</artifactId>
        <version>3.5.1</version>
        <executions>
          <execution>
            <phase>package</phase>
            <goals>
              <goal>shade</goal>
            </goals>
            <configuration>
              <!-- 调整 3：重定位 commons-lang3，避免与 Neo4j 服务端的依赖冲突 -->
              <relocations>
                <relocation>
                  <pattern>org.apache.commons.lang3</pattern>
                  <shadedPattern>sentiment.shaded.org.apache.commons.lang3</shadedPattern>
                </relocation>
              </relocations>
              <filters>
                <filter>
                  <artifact>org.neo4j:*</artifact>
                  <excludes>
                    <exclude>**</exclude>
                  </excludes>
                </filter>
              </filters>
            </configuration>
          </execution>
        </executions>
      </plugin>
    </plugins>
  </build>
</project>

package sentiment;

import com.vader.sentiment.analyzer.SentimentAnalyzer;
import com.vader.sentiment.analyzer.SentimentPolarities;
import org.neo4j.procedure.Description;
import org.neo4j.procedure.Name;
import org.neo4j.procedure.UserFunction;
import java.util.Map;

public class Sentimentable {

    @UserFunction("sentiment.score")
    @Description("Score a string with VADER. Returns compound, positive, negative, neutral.")
    public Map score(@Name("text") String text) {
        if (text == null || text.isBlank()) {
            return Map.of(
                    "compound", 0.0,
                    "positive", 0.0,
                    "negative", 0.0,
                    "neutral", 1.0
            );
        }

        final SentimentPolarities polarities = SentimentAnalyzer.getScoresFor(text);

        return Map.of(
                "compound", (double) polarities.getCompoundPolarity(),
                "positive", (double) polarities.getPositivePolarity(),
                "negative", (double) polarities.getNegativePolarity(),
                "neutral", (double) polarities.getNeutralPolarity()
        );
    }
}

package sentiment;

import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
import org.neo4j.driver.Driver;
import org.neo4j.driver.GraphDatabase;
import org.neo4j.driver.Session;
import org.neo4j.harness.Neo4j;
import org.neo4j.harness.Neo4jBuilders;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;

@TestInstance(TestInstance.Lifecycle.PER_CLASS)
public class SentimentableTest {

    private Neo4j embeddedDatabaseServer;
    private Driver driver;

    @BeforeAll
    void initializeNeo4j() {
        this.embeddedDatabaseServer = Neo4jBuilders.newInProcessBuilder()
                .withDisabledServer()
                .withFunction(Sentimentable.class)
                .build();
        this.driver = GraphDatabase.driver(embeddedDatabaseServer.boltURI());
    }

    @AfterAll
    void closeNeo4j() {
        this.driver.close();
        this.embeddedDatabaseServer.close();
    }

    @Test
    void scorePositiveSentence() {
        try (Session session = driver.session()) {
            var scores = session.run(
                    "RETURN sentiment.score('The movie was great') AS scores"
            ).single().get("scores").asMap();

            assertTrue((Double) scores.get("compound") > 0.5);
            assertTrue((Double) scores.get("positive") > 0.0);
            assertEquals(0.0, (Double) scores.get("negative"));
        }
    }

    @Test
    void capitalizationIncreasesScore() {
        try (Session session = driver.session()) {
            var normal = session.run(
                    "RETURN sentiment.score('The movie was great') AS scores"
            ).single().get("scores").asMap();

            var caps = session.run(
                    "RETURN sentiment.score('The movie was GREAT!') AS scores"
            ).single().get("scores").asMap();

            assertTrue((Double) caps.get("compound") > (Double) normal.get("compound"));
        }
    }

    @Test
    void emptyStringReturnsNeutral() {
        try (Session session = driver.session()) {
            var scores = session.run(
                    "RETURN sentiment.score('') AS scores"
            ).single().get("scores").asMap();

            assertEquals(0.0, (Double) scores.get("compound"));
            assertEquals(1.0, (Double) scores.get("neutral"));
        }
    }

    @Test
    void nullStringReturnsNeutral() {
        try (Session session = driver.session()) {
            var scores = session.run(
                    "RETURN sentiment.score(null) AS scores"
            ).single().get("scores").asMap();

            assertEquals(0.0, (Double) scores.get("compound"));
            assertEquals(1.0, (Double) scores.get("neutral"));
        }
    }
}

mvn wrapper:wrapper

./mvnw clean package

./mvnw clean package -DskipTests

cp target/sentimentable-1.0.0-SNAPSHOT.jar ~/neo4j/plugins/

docker stop neo4j-sentiment
cp target/sentimentable-1.0.0-SNAPSHOT.jar ~/neo4j/plugins/
docker start neo4j-sentiment

SHOW FUNCTIONS YIELD name WHERE name STARTS WITH 'sentiment' RETURN name;

+-----------------+
| name            |
+-----------------+
| sentiment.score |
+-----------------+

RETURN sentiment.score('The movie was great') AS scores;

{ neutral: 0.4230000078678131, negative: 0.0, positive: 0.5770000219345093, compound: 0.6248999834060669 }

RETURN sentiment.score('The movie was GREAT!') AS scores;

{ neutral: 0.36899998784065247, negative: 0.0, positive: 0.6309999823570251, compound: 0.7289999723434448 }

RETURN sentiment.score('') AS scores;

{ neutral: 1.0, negative: 0.0, positive: 0.0, compound: 0.0 }

(:Stock {symbol})-[:HAS_TICK]->(:Tick {symbol, ts, open, high, low, close, volume})
-[:HAS_HEADLINE]->(:Headline {id, symbol, ts, headline, url, publisher, compound, positive, negative, neutral})

CREATE CONSTRAINT tick_pk IF NOT EXISTS FOR (t:Tick) REQUIRE (t.symbol, t.ts) IS NODE KEY;
CREATE CONSTRAINT headline_id IF NOT EXISTS FOR (h:Headline) REQUIRE h.id IS UNIQUE;
CREATE CONSTRAINT stock_id IF NOT EXISTS FOR (s:Stock) REQUIRE s.symbol IS UNIQUE;
CREATE INDEX tick_symbol_ts IF NOT EXISTS FOR (t:Tick) ON (t.symbol, t.ts);
CREATE INDEX headline_symbol_ts IF NOT EXISTS FOR (h:Headline) ON (h.symbol, h.ts);

date,open,high,low,close,volume,Name
2013-01-02,743.98,756.93,736.15,745.68,9142645,BBRQ-FX
2013-01-03,764.41,779.16,757.72,765.16,1208771,BBRQ-FX
...

headline,url,publisher,ts,symbol
BBRQ-FX stock record revenues after analyst update,http://www.hill.net/,The Stock Chronicle,2014-10-22,BBRQ-FX
...

pip install -r requirements.txt

# data_loader.py
import pandas as pd
from neo4j import GraphDatabase
from tqdm import tqdm

URI = "bolt://localhost:7687"
AUTH = ("neo4j", "your_password_here")

TICK_CSV = "fictitious_stocks.csv"
RAW_CSV = "raw_fictitious_headlines.csv"

driver = GraphDatabase.driver(URI, auth=AUTH)

def chunks(df, size):
    for i in range(0, len(df), size):
        yield df.iloc[i:i+size].to_dict("records")

# load tick data
tick_df = (pd.read_csv(TICK_CSV)
    .dropna()
    .query("volume <= 2_147_483_647")
    .rename(columns={"date": "ts", "Name": "symbol"})
    .sort_values(["ts", "symbol"]))

tick_batches = list(chunks(tick_df, 1000))
print(f"Loading {len(tick_df):,} tick rows in {len(tick_batches)} batches...")

with driver.session() as session:
    for batch in tqdm(tick_batches, desc="Ticks", unit="batch"):
        session.run("""
            UNWIND $rows AS row
            MERGE (s:Stock {symbol: row.symbol})
            CREATE (t:Tick {
                symbol: row.symbol,
                ts: date(row.ts),
                open: row.open,
                high: row.high,
                low: row.low,
                close: row.close,
                volume: toInteger(row.volume)
            })
            CREATE (s)-[:HAS_TICK]->(t)
        """, rows=batch)

# load headlines and score at ingestion time
raw_df = pd.read_csv(RAW_CSV)
raw_batches = list(chunks(raw_df, 1000))
print(f"Loading {len(raw_df):,} headline rows in {len(raw_batches)} batches...")

with driver.session() as session:
    for batch in tqdm(raw_batches, desc="Headlines", unit="batch"):
        session.run("""
            UNWIND $rows AS row
            MATCH (s:Stock {symbol: row.symbol})
            WITH s, row, sentiment.score(row.headline) AS sc
            CREATE (h:Headline {
                id: randomUUID(),
                symbol: row.symbol,
                ts: datetime(row.ts),
                headline: row.headline,
                url: row.url,
                publisher: row.publisher,
                compound: sc.compound,
                positive: sc.positive,
                negative: sc.negative,
                neutral: sc.neutral
            })
            CREATE (s)-[:HAS_HEADLINE]->(h)
        """, rows=batch)

print("Done.")
driver.close()

python data_loader.py

MATCH (n)
CALL {
  WITH n
  DETACH DELETE n
} IN TRANSACTIONS OF 100 ROWS;

LOAD CSV WITH HEADERS FROM 'https://...' AS row
CALL {
  WITH row
  MERGE (s:Stock {symbol: row.Name})
  CREATE (t:Tick {
    symbol: row.Name,
    ts: date(row.date),
    open: toFloat(row.open),
    high: toFloat(row.high),
    low: toFloat(row.low),
    close: toFloat(row.close),
    volume: toInteger(row.volume)
  })
  CREATE (s)-[:HAS_TICK]->(t)
} IN TRANSACTIONS OF 1000 ROWS;

LOAD CSV WITH HEADERS FROM 'https://...' AS row
CALL {
  WITH row
  MATCH (s:Stock {symbol: row.symbol})
  WITH s, row, sentiment.score(row.headline) AS sc
  CREATE (h:Headline {
    id: randomUUID(),
    symbol: row.symbol,
    ts: datetime(row.ts),
    headline: row.headline,
    url: row.url,
    publisher: row.publisher,
    compound: sc.compound,
    positive: sc.positive,
    negative: sc.negative,
    neutral: sc.neutral
  })
  CREATE (s)-[:HAS_HEADLINE]->(h)
} IN TRANSACTIONS OF 1000 ROWS;

MATCH (h:Headline)
RETURN h.symbol AS symbol,
       date(h.ts) AS ts,
       left(h.headline, 30) AS headline,
       round(h.positive, 3) AS positive,
       round(h.negative, 3) AS negative,
       round(h.neutral, 3) AS neutral
ORDER BY h.symbol, h.ts
LIMIT 10;

MATCH (h:Headline)
WITH h.symbol AS symbol,
     date(h.ts) AS ts,
     avg(h.positive) AS avg_positive,
     avg(h.negative) AS avg_negative,
     avg(h.neutral) AS avg_neutral,
     count(h) AS num_headlines
RETURN symbol, ts,
       round(avg_positive, 3) AS avg_positive,
       round(avg_negative, 3) AS avg_negative,
       round(avg_neutral, 3) AS avg_neutral,
       num_headlines
ORDER BY symbol, ts
LIMIT 10;

MATCH (t:Tick)<-[:HAS_TICK]-(s:Stock)-[:HAS_HEADLINE]->(h:Headline)
WHERE date(t.ts) = date(h.ts)
RETURN t.symbol AS symbol,
       date(t.ts) AS ts,
       round(t.close, 2) AS close,
       round(h.positive, 3) AS positive,
       round(h.negative, 3) AS negative,
       round(h.neutral, 3) AS neutral
ORDER BY t.symbol, t.ts
LIMIT 10;

MATCH (h:Headline)
RETURN h.symbol,
       date(h.ts) AS ts,
       left(h.headline, 30) AS headline,
       round(h.positive, 3) AS positive
ORDER BY h.positive DESC
LIMIT 10;

MATCH (h:Headline)
RETURN h.symbol,
       date(h.ts) AS ts,
       left(h.headline, 30) AS headline,
       round(h.negative, 3) AS negative
ORDER BY h.negative DESC
LIMIT 10;

MATCH (h:Headline {symbol: 'BBRQ-FX'})
WITH h, sentiment.score(h.headline) AS live
RETURN h.symbol AS symbol,
       date(h.ts) AS ts,
       left(h.headline, 30) AS headline,
       CASE
         WHEN round(h.positive, 3) = round(live.positive, 3)
          AND round(h.negative, 3) = round(live.negative, 3)
          AND round(h.neutral, 3) = round(live.neutral, 3)
         THEN 'match'
         ELSE 'not match'
       END AS comparison
LIMIT 10;

MATCH (h:Headline)
WITH h.symbol AS symbol,
     date(h.ts) AS ts,
     avg(h.positive) AS avg_positive,
     avg(h.negative) AS avg_negative,
     avg(h.neutral) AS avg_neutral
MATCH (t:Tick {symbol: symbol})
WHERE date(t.ts) = ts
RETURN symbol,
       ts,
       round(t.close, 2) AS daily_close,
       round(avg_positive, 3) AS avg_positive,
       round(avg_negative, 3) AS avg_negative,
       round(avg_neutral, 3) AS avg_neutral
ORDER BY symbol, ts
LIMIT 10;

MATCH (h:Headline {symbol: 'BBRQ-FX'})
CALL apoc.nlp.aws.sentiment.stream(h, {
  key: $apiKey,
  secret: $apiSecret,
  nodeProperty: 'headline'
})
YIELD value
RETURN h.headline, value.sentiment, value.sentimentScore;

MATCH (h:Headline {symbol: 'BBRQ-FX'})
CALL apoc.nlp.azure.sentiment.stream(h, {
  key: $apiKey,
  url: $apiUrl,
  nodeProperty: 'headline'
})
YIELD value
RETURN h.headline, value.sentiment, value.sentimentScore;